Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add type hints to audio.py #1342

Open
wants to merge 15 commits into
base: typehints
Choose a base branch
from
3 changes: 2 additions & 1 deletion plexapi/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import os
from logging.handlers import RotatingFileHandler
from platform import uname
from typing import cast
from uuid import getnode

from plexapi.config import PlexConfig, reset_base_headers
Expand All @@ -18,7 +19,7 @@
PROJECT = 'PlexAPI'
VERSION = __version__ = const.__version__
TIMEOUT = CONFIG.get('plexapi.timeout', 30, int)
X_PLEX_CONTAINER_SIZE = CONFIG.get('plexapi.container_size', 100, int)
X_PLEX_CONTAINER_SIZE = cast(int, CONFIG.get('plexapi.container_size', 100, int))
X_PLEX_ENABLE_FAST_CONNECT = CONFIG.get('plexapi.enable_fast_connect', False, bool)

# Plex Header Configuration
Expand Down
134 changes: 86 additions & 48 deletions plexapi/audio.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,28 @@
# -*- coding: utf-8 -*-
from __future__ import annotations

import os
from pathlib import Path
from typing import TYPE_CHECKING, cast
from urllib.parse import quote_plus

from typing import Any, Dict, List, Optional, TypeVar

from plexapi import media, utils
from plexapi.base import Playable, PlexPartialObject, PlexHistory, PlexSession
from plexapi.exceptions import BadRequest
from plexapi.base import Playable, PlexHistory, PlexPartialObject, PlexSession
from plexapi.exceptions import BadRequest, PlexApiException
from plexapi.mixins import (
AdvancedSettingsMixin, SplitMergeMixin, UnmatchMatchMixin, ExtrasMixin, HubsMixin, PlayedUnplayedMixin, RatingMixin,
ArtUrlMixin, ArtMixin, PosterUrlMixin, PosterMixin, ThemeMixin, ThemeUrlMixin,
ArtistEditMixins, AlbumEditMixins, TrackEditMixins
)
from plexapi.playlist import Playlist

if TYPE_CHECKING:
from xml.etree.ElementTree import Element

from plexapi.myplex import MyPlexDevice
from plexapi.sync import SyncItem

TAudio = TypeVar("TAudio", bound="Audio")

Expand Down Expand Up @@ -53,7 +61,7 @@ class Audio(PlexPartialObject, PlayedUnplayedMixin):
"""
METADATA_TYPE = 'track'

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
self._data = data
self.addedAt = utils.toDatetime(data.attrib.get('addedAt'))
Expand Down Expand Up @@ -83,20 +91,27 @@ def _loadData(self, data):
self.userRating = utils.cast(float, data.attrib.get('userRating'))
self.viewCount = utils.cast(int, data.attrib.get('viewCount', 0))

def url(self, part):
def url(self, part: str) -> Optional[str]:
""" Returns the full URL for the audio item. Typically used for getting a specific track. """
return self._server.url(part, includeToken=True) if part else None

def _defaultSyncTitle(self):
def _defaultSyncTitle(self) -> str:
""" Returns str, default title for a new syncItem. """
return self.title
return self.title or ""

@property
def hasSonicAnalysis(self):
def hasSonicAnalysis(self) -> bool:
""" Returns True if the audio has been sonically analyzed. """
return self.musicAnalysisVersion == 1

def sync(self, bitrate, client=None, clientId=None, limit=None, title=None):
def sync(
self,
bitrate: int,
client: Optional[MyPlexDevice],
clientId: Optional[str] = None,
limit: Optional[int] = None,
title: Optional[str] = None,
) -> SyncItem:
""" Add current audio (artist, album or track) as sync item for specified device.
See :func:`~plexapi.myplex.MyPlexAccount.sync` for possible exceptions.

Expand Down Expand Up @@ -126,7 +141,7 @@ def sync(self, bitrate, client=None, clientId=None, limit=None, title=None):

section = self._server.library.sectionByID(self.librarySectionID)

sync_item.location = f'library://{section.uuid}/item/{quote_plus(self.key)}'
sync_item.location = f'library://{section.uuid}/item/{quote_plus(self.key or "")}'
sync_item.policy = Policy.create(limit)
sync_item.mediaSettings = MediaSettings.createMusic(bitrate)

Expand All @@ -136,7 +151,7 @@ def sonicallySimilar(
self: TAudio,
limit: Optional[int] = None,
maxDistance: Optional[float] = None,
**kwargs,
**kwargs: Any,
) -> List[TAudio]:
"""Returns a list of sonically similar audio items.

Expand Down Expand Up @@ -192,15 +207,15 @@ class Artist(
TAG = 'Directory'
TYPE = 'artist'

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
Audio._loadData(self, data)
self.albumSort = utils.cast(int, data.attrib.get('albumSort', '-1'))
self.collections = self.findItems(data, media.Collection)
self.countries = self.findItems(data, media.Country)
self.genres = self.findItems(data, media.Genre)
self.guids = self.findItems(data, media.Guid)
self.key = self.key.replace('/children', '') # FIX_BUG_50
self.key = self.key and self.key.replace('/children', '') # FIX_BUG_50
self.labels = self.findItems(data, media.Label)
self.locations = self.listAttrs(data, 'path', etag='Location')
self.similar = self.findItems(data, media.Similar)
Expand All @@ -211,7 +226,7 @@ def __iter__(self):
for album in self.albums():
yield album

def album(self, title):
def album(self, title: str) -> Album:
""" Returns the :class:`~plexapi.audio.Album` that matches the specified title.

Parameters:
Expand All @@ -223,15 +238,20 @@ def album(self, title):
filters={'artist.id': self.ratingKey}
)

def albums(self, **kwargs):
def albums(self, **kwargs: Any) -> List[Album]:
""" Returns a list of :class:`~plexapi.audio.Album` objects by the artist. """
return self.section().search(
libtype='album',
filters={'artist.id': self.ratingKey},
**kwargs
)

def track(self, title=None, album=None, track=None):
def track(
self,
title: Optional[str] = None,
album: Optional[str] = None,
track: Optional[int] = None,
) -> Track:
""" Returns the :class:`~plexapi.audio.Track` that matches the specified title.

Parameters:
Expand All @@ -245,20 +265,31 @@ def track(self, title=None, album=None, track=None):
key = f'{self.key}/allLeaves'
if title is not None:
return self.fetchItem(key, Track, title__iexact=title)
elif album is not None and track is not None:
if album is not None and track is not None:
return self.fetchItem(key, Track, parentTitle__iexact=album, index=track)
raise BadRequest('Missing argument: title or album and track are required')

def tracks(self, **kwargs):
def tracks(self, **kwargs: Any) -> List[Track]:
""" Returns a list of :class:`~plexapi.audio.Track` objects by the artist. """
key = f'{self.key}/allLeaves'
return self.fetchItems(key, Track, **kwargs)

def get(self, title=None, album=None, track=None):
def get(
self,
title: Optional[str] = None,
album: Optional[str] = None,
track: Optional[int] = None,
) -> Track:
""" Alias of :func:`~plexapi.audio.Artist.track`. """
return self.track(title, album, track)

def download(self, savepath=None, keep_original_name=False, subfolders=False, **kwargs):
def download(
self,
savepath: str,
keep_original_name: bool = False,
subfolders: bool = False,
**kwargs: Any,
) -> List[str]:
""" Download all tracks from the artist. See :func:`~plexapi.base.Playable.download` for details.

Parameters:
Expand All @@ -268,19 +299,21 @@ def download(self, savepath=None, keep_original_name=False, subfolders=False, **
subfolders (bool): True to separate tracks in to album folders.
**kwargs: Additional options passed into :func:`~plexapi.base.PlexObject.getStreamURL`.
"""
filepaths = []
filepaths: List[str] = []
for track in self.tracks():
if track.parentTitle is None:
raise PlexApiException(f'No parentTitle found for {track.key=} {track.title=}')
_savepath = os.path.join(savepath, track.parentTitle) if subfolders else savepath
filepaths += track.download(_savepath, keep_original_name, **kwargs)
return filepaths

def station(self):
def station(self) -> Optional[Playlist]:
""" Returns a :class:`~plexapi.playlist.Playlist` artist radio station or `None`. """
key = f'{self.key}?includeStations=1'
return next(iter(self.fetchItems(key, cls=Playlist, rtag="Stations")), None)

@property
def metadataDirectory(self):
def metadataDirectory(self) -> str:
""" Returns the Plex Media Server data directory where the metadata is stored. """
guid_hash = utils.sha1hash(self.guid)
return str(Path('Metadata') / 'Artists' / guid_hash[0] / f'{guid_hash[1:]}.bundle')
Expand Down Expand Up @@ -323,14 +356,14 @@ class Album(
TAG = 'Directory'
TYPE = 'album'

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
Audio._loadData(self, data)
self.collections = self.findItems(data, media.Collection)
self.formats = self.findItems(data, media.Format)
self.genres = self.findItems(data, media.Genre)
self.guids = self.findItems(data, media.Guid)
self.key = self.key.replace('/children', '') # FIX_BUG_50
self.key = self.key and self.key.replace('/children', '') # FIX_BUG_50
self.labels = self.findItems(data, media.Label)
self.leafCount = utils.cast(int, data.attrib.get('leafCount'))
self.loudnessAnalysisVersion = utils.cast(int, data.attrib.get('loudnessAnalysisVersion'))
Expand All @@ -352,7 +385,7 @@ def __iter__(self):
for track in self.tracks():
yield track

def track(self, title=None, track=None):
def track(self, title: Optional[str] = None, track: Optional[int] = None) -> Track:
""" Returns the :class:`~plexapi.audio.Track` that matches the specified title.

Parameters:
Expand All @@ -365,28 +398,33 @@ def track(self, title=None, track=None):
key = f'{self.key}/children'
if title is not None and not isinstance(title, int):
return self.fetchItem(key, Track, title__iexact=title)
elif track is not None or isinstance(title, int):
if track is not None or isinstance(title, int):
if isinstance(title, int):
index = title
else:
index = track
return self.fetchItem(key, Track, parentTitle__iexact=self.title, index=index)
raise BadRequest('Missing argument: title or track is required')

def tracks(self, **kwargs):
def tracks(self, **kwargs: Any) -> List[Track]:
""" Returns a list of :class:`~plexapi.audio.Track` objects in the album. """
key = f'{self.key}/children'
return self.fetchItems(key, Track, **kwargs)

def get(self, title=None, track=None):
def get(self, title: Optional[str] = None, track: Optional[int] = None) -> Track:
""" Alias of :func:`~plexapi.audio.Album.track`. """
return self.track(title, track)

def artist(self):
def artist(self) -> Optional[Artist]:
""" Return the album's :class:`~plexapi.audio.Artist`. """
return self.fetchItem(self.parentKey)

def download(self, savepath=None, keep_original_name=False, **kwargs):
return cast(Artist, self.fetchItem(self.parentKey)) if self.parentKey is not None else None

def download(
self,
savepath: str,
keep_original_name: bool = False,
**kwargs: Any,
) -> List[str]:
""" Download all tracks from the album. See :func:`~plexapi.base.Playable.download` for details.

Parameters:
Expand All @@ -395,17 +433,17 @@ def download(self, savepath=None, keep_original_name=False, **kwargs):
a friendlier filename is generated.
**kwargs: Additional options passed into :func:`~plexapi.base.PlexObject.getStreamURL`.
"""
filepaths = []
filepaths: List[str] = []
for track in self.tracks():
filepaths += track.download(savepath, keep_original_name, **kwargs)
return filepaths

def _defaultSyncTitle(self):
def _defaultSyncTitle(self) -> str:
""" Returns str, default title for a new syncItem. """
return f'{self.parentTitle} - {self.title}'

@property
def metadataDirectory(self):
def metadataDirectory(self) -> str:
""" Returns the Plex Media Server data directory where the metadata is stored. """
guid_hash = utils.sha1hash(self.guid)
return str(Path('Metadata') / 'Albums' / guid_hash[0] / f'{guid_hash[1:]}.bundle')
Expand Down Expand Up @@ -455,7 +493,7 @@ class Track(
TAG = 'Track'
TYPE = 'track'

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
Audio._loadData(self, data)
Playable._loadData(self, data)
Expand Down Expand Up @@ -487,7 +525,7 @@ def _loadData(self, data):
self.year = utils.cast(int, data.attrib.get('year'))

@property
def locations(self):
def locations(self) -> List[str]:
""" This does not exist in plex xml response but is added to have a common
interface to get the locations of the track.

Expand All @@ -497,32 +535,32 @@ def locations(self):
return [part.file for part in self.iterParts() if part]

@property
def trackNumber(self):
def trackNumber(self) -> int:
""" Returns the track number. """
return self.index

def _prettyfilename(self):
def _prettyfilename(self) -> str:
""" Returns a filename for use in download. """
return f'{self.grandparentTitle} - {self.parentTitle} - {str(self.trackNumber).zfill(2)} - {self.title}'

def album(self):
def album(self) -> Optional[Album]:
""" Return the track's :class:`~plexapi.audio.Album`. """
return self.fetchItem(self.parentKey)
return cast(Album, self.fetchItem(self.parentKey)) if self.parentKey is not None else None

def artist(self):
def artist(self) -> Optional[Artist]:
""" Return the track's :class:`~plexapi.audio.Artist`. """
return self.fetchItem(self.grandparentKey)
return cast(Artist, self.fetchItem(self.grandparentKey)) if self.grandparentKey is not None else None

def _defaultSyncTitle(self):
def _defaultSyncTitle(self) -> str:
""" Returns str, default title for a new syncItem. """
return f'{self.grandparentTitle} - {self.parentTitle} - {self.title}'

def _getWebURL(self, base=None):
def _getWebURL(self, base: Optional[str] = None) -> str:
""" Get the Plex Web URL with the correct parameters. """
return self._server._buildWebURL(base=base, endpoint='details', key=self.parentKey)

@property
def metadataDirectory(self):
def metadataDirectory(self) -> str:
""" Returns the Plex Media Server data directory where the metadata is stored. """
guid_hash = utils.sha1hash(self.parentGuid)
return str(Path('Metadata') / 'Albums' / guid_hash[0] / f'{guid_hash[1:]}.bundle')
Expand All @@ -535,7 +573,7 @@ class TrackSession(PlexSession, Track):
"""
_SESSIONTYPE = True

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
Track._loadData(self, data)
PlexSession._loadData(self, data)
Expand All @@ -548,7 +586,7 @@ class TrackHistory(PlexHistory, Track):
"""
_HISTORYTYPE = True

def _loadData(self, data):
def _loadData(self, data: Element) -> None:
""" Load attribute values from Plex XML response. """
Track._loadData(self, data)
PlexHistory._loadData(self, data)