From 0b6097c7bfc5587bc4d3e6e52775b678c56a26fc Mon Sep 17 00:00:00 2001 From: Steven Marks Date: Tue, 15 Mar 2022 23:44:06 +0000 Subject: [PATCH] feat: Adding Lidarr API (#97) * feat: Adding Lidarr API * patch: Add page & page_size const * chore: sourcery recommendations * feat: add upd_track_file --- pyarr/__init__.py | 3 +- pyarr/base.py | 122 ++++-- pyarr/const.py | 7 + pyarr/exceptions.py | 4 + pyarr/lidarr.py | 796 +++++++++++++++++++++++++++++++++++++++ pyarr/radarr.py | 7 +- pyarr/readarr.py | 27 +- pyarr/request_handler.py | 5 + pyarr/sonarr.py | 5 +- pyproject.toml | 11 +- sphinx-docs/conf.py | 17 +- 11 files changed, 936 insertions(+), 68 deletions(-) create mode 100644 pyarr/const.py create mode 100644 pyarr/lidarr.py diff --git a/pyarr/__init__.py b/pyarr/__init__.py index f32726a..274aacf 100644 --- a/pyarr/__init__.py +++ b/pyarr/__init__.py @@ -1,6 +1,7 @@ +from .lidarr import LidarrAPI from .radarr import RadarrAPI from .readarr import ReadarrAPI from .request_handler import RequestHandler from .sonarr import SonarrAPI -__all__ = ["SonarrAPI", "RadarrAPI", "RequestHandler", "ReadarrAPI"] +__all__ = ["SonarrAPI", "RadarrAPI", "RequestHandler", "ReadarrAPI", "LidarrAPI"] diff --git a/pyarr/base.py b/pyarr/base.py index 4e76d8e..3758683 100644 --- a/pyarr/base.py +++ b/pyarr/base.py @@ -1,5 +1,7 @@ from datetime import datetime +from typing import Union +from .const import PAGE, PAGE_SIZE from .request_handler import RequestHandler @@ -130,8 +132,8 @@ def get_backup(self): # GET /log def get_log( self, - page=1, - page_size=10, + page=PAGE, + page_size=PAGE_SIZE, sort_key="time", sort_dir="desc", filter_key=None, @@ -150,7 +152,6 @@ def get_log( Returns: JSON: Array """ - path = "log" params = { "page": page, "pageSize": page_size, @@ -159,11 +160,12 @@ def get_log( "filterKey": filter_key, "filterValue": filter_value, } - return self.request_get(path, self.ver_uri, params=params) + return self.request_get("log", self.ver_uri, params=params) # GET /history + # TODO: check the ID on this method may need to move to specific APIs def get_history( - self, sort_key="date", page=1, page_size=10, sort_dir="desc", id_=None + self, sort_key="date", page=PAGE, page_size=PAGE_SIZE, sort_dir="desc", id_=None ): """Gets history (grabs/failures/completed) @@ -193,8 +195,8 @@ def get_history( # GET /blocklist def get_blocklist( self, - page=1, - page_size=20, + page=PAGE, + page_size=PAGE_SIZE, sort_direction="descending", sort_key="date", ): @@ -202,7 +204,7 @@ def get_blocklist( Args: page (int, optional): Page to be returned. Defaults to 1. - page_size (int, optional): Number of results per page. Defaults to 20. + page_size (int, optional): Number of results per page. Defaults to 10. sort_direction (str, optional): Direction to sort items. Defaults to "descending". sort_key (str, optional): Field to sort by. Defaults to "date". @@ -257,7 +259,7 @@ def get_quality_profile(self, id_=None): Returns: JSON: Array """ - path = "qualityprofile" if not id_ else f"qualityprofile/{id_}" + path = f"qualityprofile/{id_}" if id_ else "qualityprofile" return self.request_get(path, self.ver_uri) # PUT /qualityprofile/{id} @@ -301,7 +303,7 @@ def get_quality_definition(self, id_=None): Returns: JSON: Array """ - path = "qualitydefinition" if not id_ else f"qualitydefinition/{id_}" + path = f"qualitydefinition/{id_}" if id_ else "qualitydefinition" return self.request_get(path, self.ver_uri) # PUT /qualitydefinition/{id} @@ -333,7 +335,7 @@ def get_indexer(self, id_=None): Returns: JSON: Array """ - path = "indexer" if not id_ else f"indexer/{id_}" + path = f"indexer/{id_}" if id_ else "indexer" return self.request_get(path, self.ver_uri) # PUT /indexer/{id} @@ -398,15 +400,18 @@ def get_task(self, id_=None): path = f"system/task/{id_}" if id_ else "system/task" return self.request_get(path, self.ver_uri) - # GET /remotePathMapping - def get_remote_path_mapping(self): - """Get a list of remote paths being mapped and used + # GET /remotepathmapping + def get_remote_path_mapping(self, id_: Union[int, None] = None): + """Get remote path mappings for downloads Directory + + Args: + id_ (Union[int, None], optional): ID for specific record. Defaults to None. Returns: JSON: Array """ - path = "remotePathMapping" - return self.request_get(path, self.ver_uri) + _path = "" if isinstance(id_, str) or id_ is None else f"/{id_}" + return self.request_get(f"remotepathmapping{_path}", self.ver_uri) # CONFIG @@ -491,18 +496,18 @@ def get_media_management(self): # NOTIFICATIONS - # GET /notification/{id} - def get_notification(self, id_=None): - """Get all notifications or a single notification by its database id + # GET /notification + def get_notification(self, id_: Union[int, None] = None): + """Get a list of all notification services, or single by ID Args: - id_ (int, optional): Notification database id. Defaults to None. + id_ (int | None, optional): Notification ID. Defaults to None. Returns: JSON: Array """ - path = "notification" if not id_ else f"notification/{id_}" - return self.request_get(path, self.ver_uri) + _path = "" if isinstance(id_, str) or id_ is None else f"/{id_}" + return self.request_get(f"notification{_path}", self.ver_uri) # GET /notification/schema def get_notification_schema(self): @@ -553,7 +558,7 @@ def get_tag(self, id_=None): Returns: JSON: Array """ - path = "tag" if not id_ else f"tag/{id_}" + path = f"tag/{id_}" if id_ else "tag" return self.request_get(path, self.ver_uri) # GET /tag/detail/{id} @@ -566,7 +571,7 @@ def get_tag_detail(self, id_=None): Returns: JSON: Array """ - path = "tag/detail" if not id_ else f"tag/detail/{id_}" + path = f"tag/detail/{id_}" if id_ else "tag/detail" return self.request_get(path, self.ver_uri) # POST /tag @@ -626,18 +631,43 @@ def get_download_client(self, id_=None): Returns: JSON: Array """ - path = "downloadclient" if not id_ else f"downloadclient/{id_}" + path = f"downloadclient/{id_}" if id_ else "downloadclient" return self.request_get(path, self.ver_uri) # GET /downloadclient/schema - def get_download_client_schema(self): - """Get a list of all the supported download clients + def get_download_client_schema(self, implementation_: Union[str, None] = None): + """Gets the schemas for the different download Clients + + Args: + implementation_ (Union[str, None], optional): Client implementation name. Defaults to None. Returns: JSON: Array """ - path = "downloadclient/schema" - return self.request_get(path, self.ver_uri) + schemas: dict = self.request_get("downloadclient/schema", self.ver_uri) + if implementation_: + return [ + schema + for schema in schemas + if schema["implementation"] == implementation_ + ] + + return schemas + + # POST /downloadclient/ + def add_download_client(self, data): + """Add a download client based on the schema information supplied + + Note: + Recommended to be used in conjunction with get_download_client_schema() + + Args: + data (dict): dictionary with download client schema and settings + + Returns: + JSON: Array + """ + return self.request_post("downloadclient", self.ver_uri, data=data) # PUT /downloadclient/{id} def upd_download_client(self, id_, data): @@ -678,9 +708,18 @@ def get_import_list(self, id_=None): Returns: JSON: Array """ - path = "importlist" if not id_ else f"importlist/{id_}" + path = f"importlist/{id_}" if id_ else "importlist" return self.request_get(path, self.ver_uri) + # POST /importlist/ + def add_import_list(self): + """This is not implemented yet + + Raises: + NotImplementedError: Error + """ + raise NotImplementedError() + # PUT /importlist/{id} def upd_import_list(self, id_, data): """Edit an importlist @@ -696,7 +735,7 @@ def upd_import_list(self, id_, data): return self.request_put(path, self.ver_uri, data=data) # DELETE /importlist/{id} - def del_import_list(self, id_): + def del_import_list(self, id_: int): """Delete an import list Args: @@ -705,5 +744,22 @@ def del_import_list(self, id_): Returns: JSON: 200 ok, 401 Unauthorized """ - path = f"importlist/{id_}" - return self.request_del(path, self.ver_uri) + return self.request_del(f"importlist/{id_}", self.ver_uri) + + # GET /config/downloadclient + def get_config_download_client(self): + """Gets download client page configuration + + Returns: + JSON: Array + """ + return self.request_get("config/downloadclient", self.ver_uri) + + # POST /notifications + def add_notifications(self): + """This is not implemented yet + + Raises: + NotImplementedError: Error + """ + raise NotImplementedError() diff --git a/pyarr/const.py b/pyarr/const.py new file mode 100644 index 0000000..7cef81c --- /dev/null +++ b/pyarr/const.py @@ -0,0 +1,7 @@ +"""PyArr Constants""" +from logging import Logger, getLogger + +LOGGER: Logger = getLogger(__package__) + +PAGE = 1 +PAGE_SIZE = 10 diff --git a/pyarr/exceptions.py b/pyarr/exceptions.py index afec58b..02fdb5c 100644 --- a/pyarr/exceptions.py +++ b/pyarr/exceptions.py @@ -24,3 +24,7 @@ class PyarrBadGateway(PyarrError): class PyarrMissingProfile(PyarrError): """Pyarr missing profile""" + + +class PyarrMethodNotAllowed(PyarrError): + """Pyarr method not allowed""" diff --git a/pyarr/lidarr.py b/pyarr/lidarr.py new file mode 100644 index 0000000..f8b230c --- /dev/null +++ b/pyarr/lidarr.py @@ -0,0 +1,796 @@ +from enum import Enum +from typing import Dict, List, Union + +from .base import BaseArrAPI +from .const import PAGE, PAGE_SIZE +from .exceptions import PyarrError, PyarrMissingProfile + + +class LidarrSortKeys(str, Enum): + """Lidarr sort keys.""" + + ALBUM_TITLE = "albums.title" + ARTIST_ID = "artistId" + DATE = "date" + DOWNLOAD_CLIENT = "downloadClient" + ID = "id" + INDEXER = "indexer" + MESSAGE = "message" + PATH = "path" + PROGRESS = "progress" + PROTOCOL = "protocol" + QUALITY = "quality" + RATINGS = "ratings" + RELEASE_DATE = "albums.releaseDate" + SOURCE_TITLE = "sourcetitle" + STATUS = "status" + TIMELEFT = "timeleft" + TITLE = "title" + + +class LidarrArtistMonitor(str, Enum): + """Lidarr Monitor types for an artist music""" + + ALL_ALBUMS = "all" + FUTURE_ALBUMS = "future" + MISSING_ALBUMS = "missing" + EXISTING_ALBUMS = "existing" + FIRST_ALBUM = "first" + LATEST_ALBUM = "latest" + + +class LidarrAPI(BaseArrAPI): + """API wrapper for Lidarr endpoints. + + Args: + RequestAPI (:obj:`str`): provides connection to API endpoint + """ + + def __init__(self, host_url: str, api_key: str): + """Initialise Lidarr API + + Args: + host_url (str): URL for Lidarr + api_key (str): API key for Lidarr + """ + + ver_uri = "/v1" + super().__init__(host_url, api_key, ver_uri) + + # POST /rootfolder + def add_root_folder( + self, + name: str, + path: str, + defaultTags: List[int], + qualityProfile: int, + metadataProfile: int, + ): + """Adds a root folder + + Args: + name (str): Name for this root folder + path (str): Location the files should be stored + defaultTags (List[int]): list of default tag IDs + qualityProfile (int): default quality profile ID + metadataProfile (int): default metadata profile ID + + Returns: + JSON: Array + """ + folder_json = { + "defaultTags": defaultTags, + "defaultQualityProfileId": qualityProfile, + "defaultMetadataProfileId": metadataProfile, + "name": name, + "path": path, + } + + return self.request_post("rootfolder", self.ver_uri, data=folder_json) + + def lookup(self, term: str): + """Search for an artist / album / song + + Args: + term (str): Search term + + Returns: + JSON: Array + """ + + return self.request_get("search", self.ver_uri, params={"term": term}) + + def get_artist(self, id_: Union[str, int, None] = None): + """Get an artist by ID or get all artists + + Args: + id_ (str | int | None, optional): Artist ID. Defaults to None. + + Returns: + JSON: Array + """ + + _path = "" if isinstance(id_, str) or id_ is None else f"/{id_}" + return self.request_get( + f"artist{_path}", + self.ver_uri, + params={"mbId": id_} if isinstance(id_, str) else None, + ) + + def _artist_json( + self, + term: str, + root_dir: str, + quality_profile_id: Union[int, None] = None, + metadata_profile_id: Union[int, None] = None, + monitored: bool = True, + artist_monitor: LidarrArtistMonitor = LidarrArtistMonitor.ALL_ALBUMS, + artist_search_for_missing_albums: bool = False, + ): + """method to help build the JSON for adding an artist + + Args: + term (str): Search term for artist + root_dir (str): root directory for music + quality_profile_id (Union[int, None], optional): Quality profile Id. Defaults to None. + metadata_profile_id (Union[int, None], optional): Metadata profile ID. Defaults to None. + monitored (bool, optional): Should this be monitored. Defaults to True. + artist_monitor (LidarrArtistMonitor, optional): should the artist be monitored. Defaults to LidarrArtistMonitor.ALL_ALBUMS. + artist_search_for_missing_albums (bool, optional): should we search for missing albums. Defaults to False. + + Raises: + PyarrMissingProfile: Raised when quality or metadata profile are missing + + Returns: + JSON: Array + """ + if quality_profile_id is None: + try: + quality_profile_id = self.get_quality_profile()[0]["id"] + except IndexError as exception: + raise PyarrMissingProfile( + "There is no Quality Profile setup" + ) from exception + if metadata_profile_id is None: + try: + metadata_profile_id = self.get_metadata_profile()[0]["id"] + except IndexError as exception: + raise PyarrMissingProfile( + "There is no Metadata Profile setup" + ) from exception + + artist = self.lookup_artist(term)[0] + artist["id"] = 0 + artist["metadataProfileId"] = metadata_profile_id + artist["qualityProfileId"] = quality_profile_id + artist["rootFolderPath"] = root_dir + artist["addOptions"] = { + "monitor": artist_monitor, + "searchForMissingAlbums": artist_search_for_missing_albums, + } + artist["monitored"] = monitored + + return artist + + def add_artist( + self, + search_term: str, + root_dir: str, + quality_profile_id: Union[int, None] = None, + metadata_profile_id: Union[int, None] = None, + monitored: bool = True, + artist_monitor: LidarrArtistMonitor = LidarrArtistMonitor.ALL_ALBUMS, + artist_search_for_missing_albums: bool = False, + ): + """Adds an artist based on a search term, must be artist name or album/single + by lidarr guid + + Args: + search_term (str): Artist name or album/single + root_dir (str): Directory for music to be stored + quality_profile_id (Union[int, None], optional): Quality profile id. Defaults to None. + metadata_profile_id (Union[int, None], optional): Metadata profile id_. Defaults to None. + monitored (bool, optional): monitor the artist. Defaults to True. + artist_monitor (LidarrArtistMonitor, optional): [description]. Defaults to LidarrArtistMonitor.ALL_ALBUMS. + artist_search_for_missing_albums (bool, optional): Search for missing albums by this artist. Defaults to False + + Returns: + JSON: Array + """ + + artist_json = self._artist_json( + search_term, + root_dir, + quality_profile_id, + metadata_profile_id, + monitored, + artist_monitor, + artist_search_for_missing_albums, + ) + return self.request_post("artist", self.ver_uri, data=artist_json) + + def upd_artist(self, data): + """Update an existing artist + + Note: + To be used in conjunction with get_artist() + + Args: + data (json): JSON data for the artist record + + Returns: + JSON: Array + """ + return self.request_put("artist", self.ver_uri, data=data) + + def delete_artist(self, id_: int): + """Delete an artist with the provided ID + + Args: + id_ (int): Artist ID to be deleted + + Returns: + None + """ + return self.request_del(f"artist/{id_}", self.ver_uri) + + def lookup_artist(self, term: str): + """Search for an Artist to add to the database + + Args: + term (str): search term to use for lookup + + Returns: + JSON: Array + """ + return self.request_get("artist/lookup", self.ver_uri, params={"term": term}) + + def get_album( + self, + albumIds: Union[int, List[int], None] = None, + artistId: Union[int, None] = None, + foreignAlbumId: Union[int, None] = None, + allArtistAlbums: bool = False, + ): + """Get a specific album by ID, or get all albums + + Args: + albumIds (int | List[int] | None, optional): database album ids. Defaults to None. + artistId (int | None, optional): database artist ids. Defaults to None. + foreignAlbumId (int | None, optional): foreign album id. Defaults to None. + allArtistAlbums (bool, optional): get all artists albums. Defaults to False. + + Returns: + JSON: Array + """ + params: Dict[str, Union[str, int, List[int]]] = { + "includeAllArtistAlbums": str(allArtistAlbums) + } + + if isinstance(albumIds, list): + params["albumids"] = albumIds + if artistId is not None: + params["artistId"] = artistId + if foreignAlbumId is not None: + params["foreignAlbumId"] = foreignAlbumId + _path = "" if isinstance(albumIds, list) or albumIds is None else f"/{albumIds}" + return self.request_get(f"album{_path}", self.ver_uri, params=params) + + def _album_json( + self, + term: str, + root_dir: str, + quality_profile_id: Union[int, None] = None, + metadata_profile_id: Union[int, None] = None, + monitored: bool = True, + artist_monitor: LidarrArtistMonitor = LidarrArtistMonitor.ALL_ALBUMS, + artist_search_for_missing_albums: bool = False, + ): + """method to help build the JSON for adding an album + + Args: + term (str): Search term for the album + root_dir (str): Director to store the album. + quality_profile_id (Union[int, None], optional): quality profile id. Defaults to None. + metadata_profile_id (Union[int, None], optional): metadata profile id. Defaults to None. + monitored (bool, optional): monitor the albums. Defaults to True. + artist_monitor (LidarrArtistMonitor, optional): monitor the artist. Defaults to LidarrArtistMonitor.ALL_ALBUMS. + artist_search_for_missing_albums (bool, optional): search for missing albums by the artist. Defaults to False. + + Raises: + PyarrMissingProfile: Error if there are no quality or metadata profiles that match + + Returns: + JSON: Array + """ + if quality_profile_id is None: + try: + quality_profile_id = self.get_quality_profile()[0]["id"] + except IndexError as exception: + raise PyarrMissingProfile( + "There is no Quality Profile setup" + ) from exception + if metadata_profile_id is None: + try: + metadata_profile_id = self.get_metadata_profile()[0]["id"] + except IndexError as exception: + raise PyarrMissingProfile( + "There is no Metadata Profile setup" + ) from exception + + artist = self.lookup_artist(term)[0] + artist["id"] = 0 + artist["metadataProfileId"] = metadata_profile_id + artist["qualityProfileId"] = quality_profile_id + artist["rootFolderPath"] = root_dir + artist["addOptions"] = { + "monitor": artist_monitor, + "searchForMissingAlbums": artist_search_for_missing_albums, + } + artist["monitored"] = monitored + + return artist + + def add_album( + self, + search_term: str, + root_dir: str, + quality_profile_id: Union[int, None] = None, + metadata_profile_id: Union[int, None] = None, + monitored: bool = True, + artist_monitor: LidarrArtistMonitor = LidarrArtistMonitor.ALL_ALBUMS, + artist_search_for_missing_albums: bool = False, + ): + """Adds an album to Lidarr + + Args: + search_term (str): name of the album to search for + root_dir (str): location to store music + quality_profile_id (Union[int, None], optional): Quality profile Id. Defaults to None. + metadata_profile_id (Union[int, None], optional): Metadata profile Id. Defaults to None. + monitored (bool, optional): should the album be monitored. Defaults to True. + artist_monitor (LidarrArtistMonitor, optional): what level to monitor the artist. Defaults to LidarrArtistMonitor.ALL_ALBUMS. + artist_search_for_missing_albums (bool, optional): search for any missing albums by this artist. Defaults to False. + + Returns: + JSON: Array + """ + album_json = self._album_json( + search_term, + root_dir, + quality_profile_id, + metadata_profile_id, + monitored, + artist_monitor, + artist_search_for_missing_albums, + ) + return self.request_post("album", self.ver_uri, data=album_json) + + def upd_album(self, data): + """Update an album + + Args: + data (json): data ti update albums + + Note: + To be used in conjunction with get_album() + + Returns: + JSON: Array + """ + return self.request_put("album", self.ver_uri, data=data) + + def delete_album(self, id_: int): + """Delete an album with the provided ID + + Args: + id_ (int): Album ID to be deleted + + Returns: + None + """ + return self.request_del(f"album/{id_}", self.ver_uri) + + def lookup_album(self, term: str): + """Search for an Album to add to the database + + Args: + term (str): search term to use for lookup + + Returns: + JSON: Array + """ + return self.request_get("album/lookup", self.ver_uri, params={"term": term}) + + # POST /command + def post_command(self): + """This function is not implemented + + Raises: + NotImplementedError: Error + """ + raise NotImplementedError("This feature is not implemented yet.") + + # GET /wanted + def get_wanted( + self, + id_: Union[int, None] = None, + sort_key: LidarrSortKeys = LidarrSortKeys.TITLE, + page: int = PAGE, + page_size: int = PAGE_SIZE, + sort_dir: str = "asc", + missing: bool = True, + ): + """Get wanted albums that are missing or not meeting cutoff + + Args: + id_ (int | None, optional): Specific album ID to return. Defaults to None. + sort_key (LidarrSortKeys, optional): id, title, ratings or quality. Defaults to LidarrSortKeys.TITLE. + page (int, optional): Page number to return. Defaults to 1. + page_size (int, optional): Number of items per page. Defaults to 10. + sort_dir (str, optional): Sort ascending or descending. Defaults to "asc". + missing (bool, optional): search for missing (True) or cutoff not met (False). Defaults to True. + + Returns: + JSON: Array + """ + params = { + "sortKey": sort_key.value, + "page": page, + "pageSize": page_size, + } + _path = "missing" if missing else "cutoff" + return self.request_get( + f"wanted/{_path}{'' if id_ is None else f'/{id_}'}", + self.ver_uri, + params=params, + ) + + # GET /parse + def get_parse(self, title: str): + """Return the music / artist with a matching filename + + Args: + title (str): file + + Returns: + JSON: Array + """ + return self.request_get("parse", self.ver_uri, params={"title": title}) + + # GET /track + def get_tracks( + self, + artistId: Union[int, None] = None, + albumId: Union[int, None] = None, + albumReleaseId: Union[int, None] = None, + trackIds: Union[int, List[int], None] = None, + ): + """Get tracks based on provided IDs + + Args: + artistId (int | None, optional): Artist ID. Defaults to None. + albumId (int | None, optional): Album ID. Defaults to None. + albumReleaseId (int | None, optional): Album Release ID. Defaults to None. + trackIds (int | list[int] | None, optional): Track IDs. Defaults to None. + + Returns: + JSON: Array + """ + params: Dict[str, Union[int, List[int]]] = {} + if artistId is not None: + params["artistId"] = artistId + if albumId is not None: + params["albumId"] = albumId + if albumReleaseId is not None: + params["albumReleaseId"] = albumReleaseId + if isinstance(trackIds, list): + params["trackIds"] = trackIds + return self.request_get( + f"track{f'/{trackIds}' if isinstance(trackIds, int) else ''}", + self.ver_uri, + params=params, + ) + + # GET /trackfile/ + def get_track_file( + self, + artistId: Union[int, None] = None, + albumId: Union[int, None] = None, + trackFileIds: Union[int, List[int], None] = None, + unmapped: bool = False, + ): + """Get track files based on IDs, or get all unmapped files + + Args: + artistId (Union[int, None], optional): Artist database ID. Defaults to None. + albumId (Union[int, None], optional): Album database ID. Defaults to None. + trackFileIds (Union[int, List[int], None], optional): specific file ids. Defaults to None. + unmapped (bool, optional): get all unmapped filterExistingFiles. Defaults to False. + + Raises: + PyarrError: where no IDs or unmapped params provided + + Returns: + JSON: Array + """ + if ( + artistId is None + and albumId is None + and trackFileIds is None + and not unmapped + ): + raise PyarrError( + "BadRequest: artistId, albumId, trackFileIds or unmapped must be provided" + ) + params: Dict[str, Union[str, int, List[int]]] = {"unmapped": str(unmapped)} + if artistId is not None: + params["artistId"] = artistId + if albumId is not None: + params["albumId"] = albumId + if isinstance(trackFileIds, list): + params["trackFileIds"] = trackFileIds + return self.request_get( + f"trackfile{f'/{trackFileIds}' if isinstance(trackFileIds, int) else ''}", + self.ver_uri, + params=params, + ) + + # PUT /trackfile/{id_} + def upd_track_file(self, data): + """Update an existing track file + + Note: + To be used in conjunction with get_track_file() + + Args: + data (json): updated data for track files + + Returns: + JSON: Array + """ + return self.request_put("trackfile", self.ver_uri, data=data) + + # DEL /trackfile/{ids_} + def delete_track_file(self, ids_: Union[int, List[int]]): + """Delete track files. Use integer for one file or list for mass deletion. + + Args: + ids_ (int | list[int]): single ID or list of IDs for files to delete + + Returns: + None + """ + return self.request_del( + f"trackfile/{'bulk' if isinstance(ids_, list) else f'{ids_}'}", + self.ver_uri, + data={"trackFileIds": ids_} if isinstance(ids_, list) else None, + ) + + # GET /metadataprofile/{id} + def get_metadata_profile(self, id_: Union[int, None] = None): + """Gets all metadata profiles or specific one with id_ + + Args: + id_ (int, optional): metadata profile id from database. Defaults to None. + + Returns: + JSON: Array + """ + _path = f"/{id_}" if id_ else "" + return self.request_get(f"metadataprofile{_path}", self.ver_uri) + + # POST /metadataprofile + def add_metadata_profile(self): + """This function is not implemented + + Raises: + NotImplementedError: Error + """ + raise NotImplementedError("This feature is not implemented yet.") + + # PUT /metadataprofile + def upd_metadata_profile(self, data): + """Update a metadata profile + + Args: + data (json): data containing metadata profile and changes + + Returns: + JSON: Array + """ + return self.request_put("metadataprofile", self.ver_uri, data=data) + + # GET /config/metadataProvider + def get_metadata_provider(self): + """Get metadata provider config (settings/metadata) + + Returns: + JSON: Array + """ + return self.request_get("config/metadataProvider", self.ver_uri) + + # PUT /config/metadataprovider + def upd_metadata_provider(self, data): + """Update metadata provider by providing json data update + + Note: + To be used in conjunction with get_metadata_provider() + + Args: + data (json): configuration data as json + + Returns: + JSON: Array + """ + return self.request_put("config/metadataProvider", self.ver_uri, data=data) + + # GET /queue + def get_queue( + self, + page: int = PAGE, + page_size: int = PAGE_SIZE, + sort_key: LidarrSortKeys = LidarrSortKeys.TIMELEFT, + unknown_artists: bool = False, + include_artist: bool = False, + include_album: bool = False, + ): + """Get the queue of download_release + + Args: + page (int, optional): Which page to load. Defaults to 1. + page_size (int, optional): Number of items per page. Defaults to 10. + sort_key (LidarrSortKeys, optional): Key to sort by. Defaults to LidarrSortKeys.TIMELEFT. + unknown_artists (bool, optional): include unknown artists. Defaults to False. + include_artist (bool, optional): Include Artists. Defaults to False. + include_album (bool, optional): Include albums. Defaults to False. + + Returns: + JSON: Array + """ + params = { + "page": page, + "pageSize": page_size, + "sortKey": sort_key, + "unknownArtists": unknown_artists, + "includeAlbum": include_album, + "includeArtist": include_artist, + } + + return self.request_get("queue", self.ver_uri, params=params) + + # GET /queue/details + def get_queue_details( + self, + artistId: Union[int, None] = None, + albumIds: Union[List[int], None] = None, + include_artist: bool = False, + include_album: bool = True, + ): + """Get queue details for artist or album + + Args: + artistId (Union[int, None], optional): Artist database ID. Defaults to None. + albumIds (Union[List[int], None], optional): Album database ID. Defaults to None. + include_artist (bool, optional): Include the artist. Defaults to False. + include_album (bool, optional): Include the album. Defaults to True. + + Returns: + JSON: Array + """ + + params: dict = { + "includeArtist": include_artist, + "includeAlbum": include_album, + } + if artistId is not None: + params["artistId"] = artistId + if albumIds is not None: + params["albumIds"] = albumIds + + return self.request_get("queue/details", self.ver_uri, params=params) + + # GET /release + def get_release( + self, artistId: Union[int, None] = None, albumId: Union[int, None] = None + ): + """Search indexers for specified fields. + + Args: + artistId (Union[int, None], optional): Artist ID from DB. Defaults to None. + albumId (Union[int, None], optional): Album IT from Database. Defaults to None. + + Returns: + JSON: Array + """ + params = {} + if artistId is not None: + params["artistId"] = artistId + if albumId is not None: + params["artistId"] = albumId + return self.request_get("release", self.ver_uri, params=params) + + # GET /rename + def get_rename(self, artistId: int, albumId: Union[int, None] = None): + """Get files matching specified id that are not properly renamed yet. + + Args: + artistId (int): Database ID for Artists + albumId (Union[int, None], optional): Album ID. Defaults to None. + + Returns: + JSON: Array + """ + params = {"artistId": artistId} + if albumId is not None: + params["albumId"] = albumId + return self.request_get( + "rename", + self.ver_uri, + params=params, + ) + + # GET /manualimport + def get_manual_import( + self, + downloadId: str, + artistId: int = 0, + folder: Union[str, None] = None, + filterExistingFiles: bool = True, + replaceExistingFiles: bool = True, + ): + """Gets a manual import list + + Args: + downloadId (str): Download IDs + artistId (int, optional): Artist Database ID. Defaults to 0. + folder (Union[str, None], optional): folder name. Defaults to None. + filterExistingFiles (bool, optional): filter files. Defaults to True. + replaceExistingFiles (bool, optional): replace files. Defaults to True. + + Returns: + JSON: Array + """ + params = { + "artistId": artistId, + "downloadId": downloadId, + "filterExistingFiles": str(filterExistingFiles), + "folder": folder if folder is not None else "", + "replaceExistingFiles": str(replaceExistingFiles), + } + return self.request_get("manualimport", self.ver_uri, params=params) + + # PUT /manualimport + def upd_manual_import(self, data): + """Update a manual import + + Note: + To be used in conjunction with get_manual_import() + + Args: + data (json): json data containing changes + + Returns: + JSON: Array + """ + return self.request_put("manualimport", self.ver_uri, data=data) + + # GET /retag + def get_retag(self, artistId: int, albumId: Union[int, None] = None): + """Get Retag + + Args: + artistId (int): ID for the artist + albumId Union[int, None], optional): ID foir the album. Defaults to None. + + Returns: + JSON: Array + """ + params = {"artistId": artistId} + if albumId is not None: + params["albumId"] = albumId + return self.request_get( + "retag", + self.ver_uri, + params=params, + ) diff --git a/pyarr/radarr.py b/pyarr/radarr.py index 740109c..384c5d5 100644 --- a/pyarr/radarr.py +++ b/pyarr/radarr.py @@ -1,4 +1,5 @@ from .base import BaseArrAPI +from .const import PAGE, PAGE_SIZE class RadarrAPI(BaseArrAPI): @@ -346,8 +347,8 @@ def get_blacklist_by_movie_id( # GET /queue def get_queue( self, - page=1, - page_size=20, + page=PAGE, + page_size=PAGE_SIZE, sort_direction="ascending", sort_key="timeLeft", include_unknown_movie_items=True, @@ -451,7 +452,7 @@ def get_indexer(self, id_=None): Returns: JSON: Array """ - path = "indexer" if not id_ else f"indexer/{id_}" + path = f"indexer/{id_}" if id_ else "indexer" return self.request_get(path, self.ver_uri) # PUT /indexer/{id} diff --git a/pyarr/readarr.py b/pyarr/readarr.py index 3109daa..43174c5 100644 --- a/pyarr/readarr.py +++ b/pyarr/readarr.py @@ -1,6 +1,6 @@ -from pyarr.exceptions import PyarrMissingProfile - from .base import BaseArrAPI +from .const import PAGE, PAGE_SIZE +from .exceptions import PyarrMissingProfile class ReadarrAPI(BaseArrAPI): @@ -65,7 +65,7 @@ def _book_json( raise PyarrMissingProfile( "There is no Metadata Profile setup" ) from exception - book = self.lookup_book(book_id_type + ":" + str(db_id))[0] + book = self.lookup_book(f"{book_id_type}:{str(db_id)}")[0] book["author"]["metadataProfileId"] = metadata_profile_id book["author"]["qualityProfileId"] = quality_profile_id @@ -172,7 +172,9 @@ def post_command(self, name, **kwargs): ## WANTED (MISSING) # GET /wanted/missing - def get_missing(self, sort_key="releaseDate", page=1, page_size=10, sort_dir="asc"): + def get_missing( + self, sort_key="releaseDate", page=PAGE, page_size=PAGE_SIZE, sort_dir="asc" + ): """Gets missing episode (episodes without files) Args: @@ -197,8 +199,8 @@ def get_missing(self, sort_key="releaseDate", page=1, page_size=10, sort_dir="as def get_cutoff( self, sort_key="releaseDate", - page=1, - page_size=10, + page=PAGE, + page_size=PAGE_SIZE, sort_dir="descending", monitored=True, ): @@ -229,8 +231,8 @@ def get_cutoff( # GET /queue def get_queue( self, - page=1, - page_size=10, + page=PAGE, + page_size=PAGE_SIZE, sort_dir="ascending", sort_key="timeleft", unknown_authors=False, @@ -274,7 +276,7 @@ def get_metadata_profile(self, id_=None): Returns: JSON: Array """ - path = "metadataprofile" if not id_ else f"metadataprofile/{id_}" + path = f"metadataprofile/{id_}" if id_ else "metadataprofile" return self.request_get(path, self.ver_uri) # GET /delayprofile/{id} @@ -287,7 +289,7 @@ def get_delay_profile(self, id_): Returns: JSON: Array """ - path = "delayprofile" if not id_ else f"delayprofile/{id_}" + path = f"delayprofile/{id_}" if id_ else "delayprofile" return self.request_get(path, self.ver_uri) # GET /releaseprofile/{id} @@ -300,7 +302,7 @@ def get_release_profile(self, id_=None): Returns: JSON: Array """ - path = "releaseprofile" if not id_ else f"releaseprofile/{id_}" + path = f"releaseprofile/{id_}" if id_ else "releaseprofile" return self.request_get(path, self.ver_uri) ## BOOKS @@ -332,9 +334,8 @@ def lookup_book(self, term): Returns: JSON: Array """ - params = {"term": term} path = "book/lookup" - return self.request_get(path, self.ver_uri, params=params) + return self.request_get(path, self.ver_uri, params={"term": term}) # POST /book def add_book( diff --git a/pyarr/request_handler.py b/pyarr/request_handler.py index c8053d3..85f9b7b 100644 --- a/pyarr/request_handler.py +++ b/pyarr/request_handler.py @@ -4,6 +4,7 @@ PyarrAccessRestricted, PyarrBadGateway, PyarrConnectionError, + PyarrMethodNotAllowed, PyarrResourceNotFound, PyarrUnauthorizedError, ) @@ -76,6 +77,7 @@ def request_get(self, path, ver_uri="", params=None): raise PyarrConnectionError( "Timeout occurred while connecting to API" ) from exception + return _process_response(res) def request_post(self, path, ver_uri="", params=None, data=None): @@ -184,6 +186,9 @@ def _process_response(res): raise PyarrResourceNotFound("Resource not found") if res.status_code == 502: raise PyarrBadGateway("Bad Gateway. Check your server is accessible") + if res.status_code == 405: + raise PyarrMethodNotAllowed(f"The endpoint {res.url} is not allowed") + content_type = res.headers.get("Content-Type", "") if "application/json" in content_type: return res.json() diff --git a/pyarr/sonarr.py b/pyarr/sonarr.py index e4667ce..b8cf080 100644 --- a/pyarr/sonarr.py +++ b/pyarr/sonarr.py @@ -1,4 +1,5 @@ from .base import BaseArrAPI +from .const import PAGE, PAGE_SIZE class SonarrAPI(BaseArrAPI): @@ -215,7 +216,9 @@ def upd_episode_file_quality(self, id_, data): path = f"episodefile/{id_}" return self.request_put(path, self.ver_uri, data=data) - def get_wanted(self, sort_key="airDateUtc", page=1, page_size=10, sort_dir="asc"): + def get_wanted( + self, sort_key="airDateUtc", page=PAGE, page_size=PAGE_SIZE, sort_dir="asc" + ): """Gets missing episode (episodes without files) Args: diff --git a/pyproject.toml b/pyproject.toml index 9cdd0f6..937b99d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "pyarr" -version = "3.0.1" +version = "3.1.0" description = "Python client for Servarr API's (Sonarr, Radarr, Readarr)" authors = ["Steven Marks "] license = "MIT" @@ -74,6 +74,15 @@ disallow_untyped_defs = true [tool.isort] profile = "black" +# will group `import x` and `from x import` of the same module. +force_sort_within_sections = true +known_first_party = [ + "pyarr", + "tests", +] +forced_separate = [ + "tests", +] [tool.interrogate] ignore-init-method = true diff --git a/sphinx-docs/conf.py b/sphinx-docs/conf.py index 5bb8034..de4226f 100644 --- a/sphinx-docs/conf.py +++ b/sphinx-docs/conf.py @@ -1,8 +1,8 @@ # -*- coding: utf-8 -*- +from os.path import abspath, dirname, join import re import sys -from os.path import abspath, dirname, join path = dirname(dirname(abspath(__file__))) sys.path.append(path) @@ -65,21 +65,6 @@ # -- Options for LaTeX output ------------------------------------------------ -latex_elements = { - # The paper size ('letterpaper' or 'a4paper'). - # - # 'papersize': 'letterpaper', - # The font size ('10pt', '11pt' or '12pt'). - # - # 'pointsize': '10pt', - # Additional stuff for the LaTeX preamble. - # - # 'preamble': '', - # Latex figure (float) alignment - # - # 'figure_align': 'htbp', -} - latex_documents = [ ("index", "{0}.tex".format(slug), project, author, "manual"), ]