feat: type Plex ingestion pipeline

docker/pyproject.deps.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mcp-plex"
-version = "1.0.2"
+version = "1.0.3"
 requires-python = ">=3.11,<3.13"
 dependencies = [
   "fastmcp>=2.11.2",

mcp_plex/loader/pipeline/channels.py

@@ -17,8 +17,10 @@
 
 try:  # Only import plexapi when available; the sample data mode does not require it.
     from plexapi.base import PlexPartialObject
+    from plexapi.video import Episode, Movie, Show
 except Exception:
     PlexPartialObject = object  # type: ignore[assignment]
+    Episode = Movie = Show = PlexPartialObject  # type: ignore[assignment]
 
 T = TypeVar("T")
 
@@ -46,15 +48,15 @@
 class MovieBatch:
     """Batch of Plex movie items pending metadata enrichment."""
 
-    movies: list["PlexPartialObject"]
+    movies: list["Movie"]
 
 
 @dataclass(slots=True)
 class EpisodeBatch:
     """Batch of Plex episodes along with their parent show."""
 
-    show: "PlexPartialObject"
-    episodes: list["PlexPartialObject"]
+    show: "Show"
+    episodes: list["Episode"]
 
 
 @dataclass(slots=True)

mcp_plex/loader/pipeline/ingestion.py

@@ -9,7 +9,7 @@
 
 import asyncio
 import logging
-from typing import Sequence
+from typing import TYPE_CHECKING, Sequence
 
 from ...common.types import AggregatedItem
 from .channels import (
@@ -20,14 +20,24 @@
     chunk_sequence,
 )
 
+if TYPE_CHECKING:  # pragma: no cover - imported for typing
+    from plexapi.server import PlexServer
+    from plexapi.video import Episode, Movie, Season, Show
+else:  # pragma: no cover - runtime import with graceful fallback
+    try:
+        from plexapi.server import PlexServer
+        from plexapi.video import Episode, Movie, Season, Show
+    except Exception:  # pragma: no cover - plexapi optional at runtime
+        PlexServer = Movie = Show = Season = Episode = object  # type: ignore[assignment]
+
 
 class IngestionStage:
     """Coordinate ingesting items from Plex or bundled sample data."""
 
     def __init__(
         self,
         *,
-        plex_server: object | None,
+        plex_server: PlexServer | None,
         sample_items: Sequence[AggregatedItem] | None,
         movie_batch_size: int,
         episode_batch_size: int,
@@ -145,17 +155,18 @@ async def _run_plex_ingestion(self) -> None:
     async def _ingest_plex(
         self,
         *,
-        plex_server: object,
+        plex_server: PlexServer,
         movie_batch_size: int,
         episode_batch_size: int,
         output_queue: IngestQueue,
         logger: logging.Logger,
     ) -> None:
         """Retrieve Plex media and place batches onto *output_queue*."""
 
-        movies_attr = getattr(plex_server, "movies", [])
-        movies_source = movies_attr() if callable(movies_attr) else movies_attr
-        movies = list(movies_source)
+        library = plex_server.library
+
+        movies_section = library.section("Movies")
+        movies: list[Movie] = list(movies_section.all())
         logger.info(
             "Discovered %d Plex movie(s) for ingestion.",
             len(movies),
@@ -180,9 +191,8 @@ async def _ingest_plex(
                 self._items_ingested,
             )
 
-        shows_attr = getattr(plex_server, "shows", [])
-        shows_source = shows_attr() if callable(shows_attr) else shows_attr
-        shows = list(shows_source)
+        shows_section = library.section("TV Shows")
+        shows: list[Show] = list(shows_section.all())
         logger.info(
             "Discovered %d Plex show(s) for ingestion.",
             len(shows),
@@ -191,11 +201,10 @@ async def _ingest_plex(
         episode_total = 0
         for show in shows:
             show_title = getattr(show, "title", str(show))
-            episodes_attr = getattr(show, "episodes", [])
-            episodes_source = (
-                episodes_attr() if callable(episodes_attr) else episodes_attr
-            )
-            episodes = list(episodes_source)
+            seasons: list[Season] = list(show.seasons())
+            episodes: list[Episode] = []
+            for season in seasons:
+                episodes.extend(season.episodes())
             if not episodes:
                 logger.debug("Show %s yielded no episodes for ingestion.", show_title)
             for batch_index, chunk in enumerate(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mcp-plex"
-version = "1.0.2"
+version = "1.0.3"
 
 description = "Plex-Oriented Model Context Protocol Server"
 requires-python = ">=3.11,<3.13"

tests/test_ingestion_stage.py

@@ -1,7 +1,11 @@
 import asyncio
 import logging
+from typing import cast
+from unittest.mock import Mock, create_autospec
 
 import pytest
+from plexapi.server import PlexServer
+from plexapi.video import Episode, Movie, Season, Show
 
 from mcp_plex.common.types import AggregatedItem, PlexItem
 from mcp_plex.loader.pipeline.channels import (
@@ -28,7 +32,7 @@ def test_ingestion_stage_logger_name() -> None:
     async def scenario() -> str:
         queue: asyncio.Queue = asyncio.Queue()
         stage = IngestionStage(
-            plex_server=object(),
+            plex_server=cast(PlexServer, object()),
             sample_items=None,
             movie_batch_size=50,
             episode_batch_size=25,
@@ -159,45 +163,49 @@ async def scenario() -> tuple[list[SampleBatch], object | None, object | None, i
 def test_ingestion_stage_ingest_plex_batches_movies_and_episodes(caplog) -> None:
     caplog.set_level(logging.INFO)
 
-    class FakeMovie:
-        def __init__(self, title: str) -> None:
-            self.title = title
+    sentinel = object()
 
-    class FakeEpisode:
-        def __init__(self, title: str) -> None:
-            self.title = title
+    async def scenario() -> tuple[list[object], int, int, Mock]:
+        queue: asyncio.Queue = asyncio.Queue()
 
-    class FakeShow:
-        def __init__(self, title: str, episode_titles: list[str]) -> None:
-            self.title = title
-            self._episodes = [FakeEpisode(ep_title) for ep_title in episode_titles]
+        movie_section = Mock()
+        movies = [
+            create_autospec(Movie, instance=True, title="Movie 1"),
+            create_autospec(Movie, instance=True, title="Movie 2"),
+            create_autospec(Movie, instance=True, title="Movie 3"),
+        ]
+        movie_section.all.return_value = movies
 
-        def episodes(self) -> list[FakeEpisode]:
-            return list(self._episodes)
+        def _episodes(titles: list[str]) -> list[Episode]:
+            return [create_autospec(Episode, instance=True, title=title) for title in titles]
 
-    class FakePlex:
-        def __init__(self) -> None:
-            self._movies = [
-                FakeMovie("Movie 1"),
-                FakeMovie("Movie 2"),
-                FakeMovie("Movie 3"),
-            ]
-            self._shows = [
-                FakeShow("Show A", ["S01E01", "S01E02", "S01E03"]),
-                FakeShow("Show B", ["S01E01", "S01E02"]),
-            ]
+        show_a_season_1 = create_autospec(Season, instance=True)
+        show_a_season_1.episodes.return_value = _episodes(["S01E01", "S01E02"])
+        show_a_season_2 = create_autospec(Season, instance=True)
+        show_a_season_2.episodes.return_value = _episodes(["S01E03"])
 
-        def movies(self) -> list[FakeMovie]:
-            return list(self._movies)
+        show_a = create_autospec(Show, instance=True, title="Show A")
+        show_a.seasons.return_value = [show_a_season_1, show_a_season_2]
 
-        def shows(self) -> list[FakeShow]:
-            return list(self._shows)
+        show_b_season_1 = create_autospec(Season, instance=True)
+        show_b_season_1.episodes.return_value = _episodes(["S01E01", "S01E02"])
 
-    sentinel = object()
+        show_b = create_autospec(Show, instance=True, title="Show B")
+        show_b.seasons.return_value = [show_b_season_1]
+
+        shows = [show_a, show_b]
+        show_section = Mock()
+        show_section.all.return_value = shows
+
+        library = Mock()
+        library.section.side_effect = lambda name: {
+            "Movies": movie_section,
+            "TV Shows": show_section,
+        }[name]
+
+        plex = create_autospec(PlexServer, instance=True)
+        plex.library = library
 
-    async def scenario() -> tuple[list[object], int, int]:
-        queue: asyncio.Queue = asyncio.Queue()
-        plex = FakePlex()
         stage = IngestionStage(
             plex_server=plex,
             sample_items=None,
@@ -220,10 +228,14 @@ async def scenario() -> tuple[list[object], int, int]:
         while not queue.empty():
             batches.append(await queue.get())
 
-        return batches, stage.items_ingested, stage.batches_ingested
+        return batches, stage.items_ingested, stage.batches_ingested, library
 
-    batches, items_ingested, batches_ingested = asyncio.run(scenario())
+    batches, items_ingested, batches_ingested, library = asyncio.run(scenario())
 
+    assert library.section.call_args_list == [
+        (("Movies",),),
+        (("TV Shows",),),
+    ]
     assert items_ingested == 8
     assert batches_ingested == 5