Replace python-plexapi with custom Json service

Tidy ups, fixes and improvements

Author: Buried-In-Code

mediux_posters/__main__.py

@@ -5,7 +5,6 @@
 from platform import python_version
 from typing import Annotated, Protocol, TypeVar
 
-from plexapi.exceptions import Unauthorized
 from typer import Abort, Context, Exit, Option, Typer
 
 from mediux_posters import __version__, get_cache_root, setup_logging
@@ -63,11 +62,8 @@ def setup(
         raise Abort
     mediux = Mediux(base_url=settings.mediux.base_url, api_key=settings.mediux.api_key)
     service_list = []
-    try:
-        if settings.plex.token:
-            service_list.append(Plex(base_url=settings.plex.base_url, token=settings.plex.token))
-    except Unauthorized as err:
-        LOGGER.warning(err)
+    if settings.plex.token:
+        service_list.append(Plex(base_url=settings.plex.base_url, token=settings.plex.token))
     if settings.jellyfin.token:
         service_list.append(
             Jellyfin(base_url=settings.jellyfin.base_url, token=settings.jellyfin.token)
@@ -129,14 +125,16 @@ def process_image(
             setattr(entry, attribute, True)
         else:
             if should_log:
-                LOGGER.info("Downloading '%s' by '%s'", set_data.set_title, set_data.username)
+                LOGGER.info(
+                    "[Mediux] Downloading '%s' by '%s'", set_data.set_title, set_data.username
+                )
                 should_log = False
             mediux.download_image(file_id=file_info.id, output=image_file)
             setattr(
                 entry,
                 attribute,
                 service.upload_image(
-                    obj=entry, image_file=image_file, kometa_integration=kometa_integration
+                    object_id=entry.id, image_file=image_file, kometa_integration=kometa_integration
                 ),
             )
     return should_log
@@ -152,7 +150,9 @@ def process_set_data(
 ) -> bool:
     should_log = True
     if entry.all_posters_uploaded:
-        LOGGER.info("All posters have been uploaded, skipping remaining sets")
+        LOGGER.info(
+            "[%s] All posters have been uploaded, skipping remaining sets", type(service).__name__
+        )
         return False
 
     should_log = process_image(
@@ -180,7 +180,8 @@ def process_set_data(
         kometa_integration=kometa_integration,
     )
     if media_type is MediaType.SHOW:
-        for season in entry.seasons:
+        for season in service.list_seasons(show_id=entry.id):
+            entry.seasons.append(season)
             mediux_season = next(
                 (x for x in set_data.show.seasons if x.number == season.number), None
             )
@@ -198,7 +199,8 @@ def process_set_data(
                 service=service,
                 kometa_integration=kometa_integration,
             )
-            for episode in season.episodes:
+            for episode in service.list_episodes(show_id=entry.id, season_id=season.id):
+                season.episodes.append(episode)
                 mediux_episode = next(
                     (x for x in mediux_season.episodes if x.number == episode.number), None
                 )
@@ -217,7 +219,8 @@ def process_set_data(
                     kometa_integration=kometa_integration,
                 )
     elif media_type is MediaType.COLLECTION:
-        for movie in entry.movies:
+        for movie in service.list_collection_movies(collection_id=entry.id):
+            entry.movies.append(movie)
             mediux_movie = next(
                 (x for x in set_data.collection.movies if x.tmdb_id == movie.tmdb_id), None
             )
@@ -483,7 +486,7 @@ def media_posters(
                 style="subtitle",
             )
             if simple_clean:
-                LOGGER.info("Cleaning %s cache", entry.display_name)
+                LOGGER.info("Cleaning %s from cache", entry.display_name)
                 delete_folder(folder=get_cached_image(slugify(value=entry.display_name)))
                 if media_type is MediaType.COLLECTION:
                     for movie in entry.movies:
@@ -635,7 +638,7 @@ def set_posters(
                 LOGGER.warning("[%s] %s", type(service).__name__, err)
                 break
             if simple_clean:
-                LOGGER.info("Cleaning %s cache", entry.display_name)
+                LOGGER.info("Cleaning %s from cache", entry.display_name)
                 delete_folder(folder=get_cached_image(slugify(value=entry.display_name)))
                 if media_type is MediaType.COLLECTION:
                     for movie in entry.movies:

mediux_posters/mediux/service.py

@@ -19,7 +19,9 @@
 from mediux_posters.utils import MediaType
 
 LOGGER = logging.getLogger(__name__)
-MINUTE = 60
+# 60 Calls per Minute
+CALLS = 60
+PERIOD = 60
 
 
 class Mediux:
@@ -105,7 +107,7 @@ def __init__(self, base_url: str, api_key: str):
         )
 
     @sleep_and_retry
-    @limits(calls=30, period=MINUTE)
+    @limits(calls=CALLS, period=PERIOD)
     def _perform_graphql_request(self, query: str) -> dict[str, Any]:
         try:
             response = self.client.post("/graphql", json={"query": query})
@@ -318,6 +320,8 @@ def get_set(
             else None
         )
 
+    @sleep_and_retry
+    @limits(calls=CALLS, period=PERIOD)
     def download_image(self, file_id: str, output: Path) -> None:
         output.parent.mkdir(parents=True, exist_ok=True)
         output.unlink(missing_ok=True)

mediux_posters/services/_base/service.py

@@ -21,6 +21,12 @@
 
 
 class BaseService(ABC, Generic[T, S, E, C, M]):
+    @abstractmethod
+    def list_episodes(self, show_id: int | str, season_id: int | str) -> list[E]: ...
+
+    @abstractmethod
+    def list_seasons(self, show_id: int | str) -> list[S]: ...
+
     @abstractmethod
     def list_shows(self, skip_libraries: list[str] | None = None) -> list[T]: ...
 
@@ -33,6 +39,9 @@ def list_collections(self, skip_libraries: list[str] | None = None) -> list[C]:
     @abstractmethod
     def get_collection(self, tmdb_id: int) -> C | None: ...
 
+    @abstractmethod
+    def list_collection_movies(self, collection_id: int | str) -> list[M]: ...
+
     @abstractmethod
     def list_movies(self, skip_libraries: list[str] | None = None) -> list[M]: ...
 
@@ -66,5 +75,5 @@ def get(self, media_type: MediaType, tmdb_id: int) -> T | C | M | None:
 
     @abstractmethod
     def upload_image(
-        self, obj: T | S | E | M | C, image_file: Path, kometa_integration: bool
+        self, object_id: int | str, image_file: Path, kometa_integration: bool
     ) -> bool: ...

mediux_posters/services/jellyfin/service.py

@@ -10,7 +10,6 @@
 
 from httpx import Client, HTTPStatusError, RequestError, TimeoutException
 from pydantic import TypeAdapter, ValidationError
-from ratelimit import limits, sleep_and_retry
 
 from mediux_posters import __version__
 from mediux_posters.console import CONSOLE
@@ -26,7 +25,6 @@
 )
 
 LOGGER = logging.getLogger(__name__)
-MINUTE = 60
 
 
 class Jellyfin(BaseService[Show, Season, Episode, Collection, Movie]):
@@ -44,8 +42,6 @@ def __init__(self, base_url: str, token: str):
     def extract_id(cls, entry: dict, prefix: str = "Tmdb") -> str | None:
         return entry.get("ProviderIds", {}).get(prefix)
 
-    @sleep_and_retry
-    @limits(calls=30, period=MINUTE)
     def _perform_get_request(
         self, endpoint: str, params: dict[str, str | list[str]] | None = None
     ) -> dict:
@@ -71,8 +67,6 @@ def _perform_get_request(
         except TimeoutException as err:
             raise ServiceError("Service took too long to respond") from err
 
-    @sleep_and_retry
-    @limits(calls=30, period=MINUTE)
     def _perform_post_request(
         self, endpoint: str, body: bytes, headers: dict[str, str] | None = None
     ) -> None:
@@ -114,15 +108,18 @@ def _list_libraries(
         except ValidationError as err:
             raise ServiceError(err) from err
 
-    def _parse_show(self, jellyfin_show: dict) -> Show:
-        show = TypeAdapter(Show).validate_python(jellyfin_show)
-        for jellyfin_season in self._list_seasons(show_id=show.id):
-            season = TypeAdapter(Season).validate_python(jellyfin_season)
-            for jellyfin_episode in self._list_episodes(show_id=show.id, season_id=season.id):
-                episode = TypeAdapter(Episode).validate_python(jellyfin_episode)
-                season.episodes.append(episode)
-            show.seasons.append(season)
-        return show
+    def list_episodes(self, show_id: str, season_id: str) -> list[Episode]:
+        results = self._perform_get_request(
+            endpoint=f"/Shows/{show_id}/Episodes",
+            params={"seasonId": season_id, "fields": ["ProviderIds"]},
+        ).get("Items", [])
+        return TypeAdapter(list[Episode]).validate_python(results)
+
+    def list_seasons(self, show_id: str) -> list[Season]:
+        results = self._perform_get_request(
+            endpoint=f"/Shows/{show_id}/Seasons", params={"fields": ["ProviderIds"]}
+        ).get("Items", [])
+        return TypeAdapter(list[Season]).validate_python(results)
 
     def _list_shows(
         self,
@@ -149,48 +146,26 @@ def _list_shows(
                 if not tmdb or (tmdb_id is not None and int(tmdb) != tmdb_id):
                     continue
                 try:
-                    output.append(self._parse_show(jellyfin_show=result))
+                    result = TypeAdapter(Show).validate_python(result)
+                    output.append(result)
                 except ValidationError as err:
                     raise ServiceError(err) from err
         return output
 
-    def _list_seasons(self, show_id: str) -> list[dict]:
-        return self._perform_get_request(
-            endpoint=f"/Shows/{show_id}/Seasons", params={"fields": ["ProviderIds"]}
-        ).get("Items", [])
-
-    def _list_episodes(self, show_id: str, season_id: str) -> list[dict]:
-        return self._perform_get_request(
-            endpoint=f"/Shows/{show_id}/Episodes",
-            params={"seasonId": season_id, "fields": ["ProviderIds"]},
-        ).get("Items", [])
-
     def list_shows(self, skip_libraries: list[str] | None = None) -> list[Show]:
         return self._list_shows(skip_libraries=skip_libraries)
 
     def get_show(self, tmdb_id: int) -> Show | None:
         return next(iter(self._list_shows(tmdb_id=tmdb_id)), None)
 
-    def _list_collections(
-        self,
-        skip_libraries: list[str] | None = None,
-        tmdb_id: int | None = None,  # noqa: ARG002
-        collection_id: str | None = None,  # noqa: ARG002
-    ) -> list[Collection]:
-        libraries = self._list_libraries(media_type="unknown", skip_libraries=skip_libraries)
-        output = []
-        for _library in libraries:
-            pass
-        return output
-
-    def list_collections(self, skip_libraries: list[str] | None = None) -> list[Collection]:
-        return self._list_collections(skip_libraries=skip_libraries)
+    def list_collections(self, skip_libraries: list[str] | None = None) -> list[Collection]:  # noqa: ARG002
+        return []
 
-    def get_collection(self, tmdb_id: int) -> Collection | None:
-        return next(iter(self._list_collections(tmdb_id=tmdb_id)), None)
+    def get_collection(self, tmdb_id: int) -> Collection | None:  # noqa: ARG002
+        return None
 
-    def _parse_movie(self, jellyfin_movie: dict) -> Movie:
-        return TypeAdapter(Movie).validate_python(jellyfin_movie)
+    def list_collection_movies(self, collection_id: int | str) -> list[Movie]:  # noqa: ARG002
+        return []
 
     def _list_movies(
         self,
@@ -217,7 +192,8 @@ def _list_movies(
                 if not tmdb or (tmdb_id is not None and int(tmdb) != tmdb_id):
                     continue
                 try:
-                    output.append(self._parse_movie(jellyfin_movie=result))
+                    result = TypeAdapter(Movie).validate_python(result)
+                    output.append(result)
                 except ValidationError as err:
                     raise ServiceError(err) from err
         return output
@@ -230,7 +206,7 @@ def get_movie(self, tmdb_id: int) -> Movie | None:
 
     def upload_image(
         self,
-        obj: Show | Season | Episode | Movie | Collection,
+        object_id: int | str,
         image_file: Path,
         kometa_integration: bool,  # noqa: ARG002
     ) -> bool:
@@ -244,7 +220,7 @@ def upload_image(
                 image_data = b64encode(stream.read())
             try:
                 self._perform_post_request(
-                    endpoint=f"/Items/{obj.id}/Images/{image_type}",
+                    endpoint=f"/Items/{object_id}/Images/{image_type}",
                     headers=headers,
                     body=image_data,
                 )