francwa/alfred
1
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor(tmdb): ACL pass — push VOs into DTOs, split search per media type

Browse Source 
Anti-corruption boundary tightened on the TMDB adapter:

* TmdbMovieInfo / TmdbShowInfo now carry domain VOs (TmdbId, ImdbId,
  MovieTitle, ReleaseYear, ShowStatus) instead of raw scalars —
  validation happens at the boundary, not three layers later.
* ShowStatus enum added (domain/tv_shows/value_objects) with a
  from_tmdb() mapper that falls back to UNKNOWN + logs a warning on
  unrecognized values. TVShow.status is now ShowStatus, not str.
* MovieTitle cap raised from 100 to 150 chars.
* MediaResult / ExternalIds dropped. Replaced by per-media search
  DTOs: TmdbMovieSearchResult and TmdbShowSearchResult. Neither
  carries imdb_id — search no longer enriches with external_ids
  (callers needing imdb_id follow up with get_movie_info /
  get_tv_show_info on the chosen tmdb_id).
* TMDBClient: search_multi / search_media / _parse_result removed.
  search_movies (/search/movie) and search_shows (/search/tv) added,
  each parsing hits into VO-typed DTOs.
* SearchMovieUseCase returns a list of MovieHit (flattened to
  primitives for the agent). New symmetric SearchShowUseCase +
  ShowHit / SearchShowResponse DTOs.
* agent/tools/api.py: find_media_imdb_id → search_movies +
  search_shows wrappers.
* FileEntry moved from domain/shared/ports/filesystem_scanner.py to
  domain/shared/file_entry.py (it's a DTO, not a Protocol); size_kb
  (float) → size (int bytes). Scanner and SubtitleIdentifier
  updated.

Tests: 79/79 pass on tests/infrastructure/api/ +
tests/application/test_search_movie.py +
tests/application/test_search_show.py.
This commit is contained in:
Francwa
2026-05-26 05:45:30 +02:00
parent cffafa2e60
commit c62ae81275
29 changed files with 735 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alfred/agent/tools/api.py
+24 -12
View File
@@ -5,6 +5,7 @@ from typing import Any
from alfred.application.movies import SearchMovieUseCase
from alfred.application.torrents import AddTorrentUseCase, SearchTorrentsUseCase
from alfred.application.tv_shows import SearchShowUseCase
from alfred.infrastructure.api.knaben import knaben_client
from alfred.infrastructure.api.qbittorrent import qbittorrent_client
from alfred.infrastructure.api.tmdb import tmdb_client
@@ -13,25 +14,36 @@ from alfred.infrastructure.persistence import get_memory
logger = logging.getLogger(__name__)
def find_media_imdb_id(media_title: str) -> dict[str, Any]:
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/find_media_imdb_id.yaml."""
def search_movies(media_title: str) -> dict[str, Any]:
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/search_movies.yaml."""
use_case = SearchMovieUseCase(tmdb_client)
response = use_case.execute(media_title)
result = response.to_dict()
if result.get("status") == "ok":
memory = get_memory()
memory.stm.set_entity(
"last_media_search",
{
"title": result.get("title"),
"imdb_id": result.get("imdb_id"),
"media_type": result.get("media_type"),
"tmdb_id": result.get("tmdb_id"),
},
memory.stm.set_entity("last_movie_search", {"hits": result.get("hits", [])})
memory.stm.set_topic("searching_movie")
logger.debug(
f"Stored movie search result in STM: {len(result.get('hits', []))} hits"
)
return result
def search_shows(show_title: str) -> dict[str, Any]:
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/search_shows.yaml."""
use_case = SearchShowUseCase(tmdb_client)
response = use_case.execute(show_title)
result = response.to_dict()
if result.get("status") == "ok":
memory = get_memory()
memory.stm.set_entity("last_show_search", {"hits": result.get("hits", [])})
memory.stm.set_topic("searching_show")
logger.debug(
f"Stored show search result in STM: {len(result.get('hits', []))} hits"
)
memory.stm.set_topic("searching_media")
logger.debug(f"Stored media search result in STM: {result.get('title')}")
return result
alfred/application/filesystem/__init__.py
-4
View File
@@ -7,7 +7,6 @@ from .dto import (
ManageSubtitlesResponse,
MoveMediaResponse,
PlacedSubtitle,
SetFolderPathResponse,
)
from .list_folder import ListFolderUseCase
from .manage_subtitles import ManageSubtitlesUseCase
@@ -22,10 +21,8 @@ from .resolve_destination import (
resolve_season_destination,
resolve_series_destination,
)
from .set_folder_path import SetFolderPathUseCase
__all__ = [
"SetFolderPathUseCase",
"ListFolderUseCase",
"CreateSeedLinksUseCase",
"MoveMediaUseCase",
@@ -38,7 +35,6 @@ __all__ = [
"resolve_episode_destination",
"resolve_movie_destination",
"resolve_series_destination",
"SetFolderPathResponse",
"ListFolderResponse",
"CreateSeedLinksResponse",
"MoveMediaResponse",
alfred/application/movies/__init__.py
+2 -4
View File
@@ -1,12 +1,10 @@
"""Movie use cases."""
from .dto import SearchMovieResponse
from .rescan import MovieRescanFailed, rescan_movie
from .dto import MovieHit, SearchMovieResponse
from .search_movie import SearchMovieUseCase
__all__ = [
"MovieRescanFailed",
"MovieHit",
"SearchMovieResponse",
"SearchMovieUseCase",
"rescan_movie",
]
alfred/application/movies/dto.py
+19 -17
View File
@@ -1,6 +1,21 @@
"""Movie application DTOs."""
from dataclasses import dataclass
from dataclasses import dataclass, field
@dataclass(frozen=True)
class MovieHit:
"""One movie hit, flattened for transport to the agent."""
tmdb_id: int
title: str
release_year: int | None = None
def to_dict(self) -> dict:
out: dict = {"tmdb_id": self.tmdb_id, "title": self.title}
if self.release_year is not None:
out["release_year"] = self.release_year
return out
@dataclass
@@ -8,31 +23,18 @@ class SearchMovieResponse:
"""Response from searching for a movie."""
status: str
imdb_id: str | None = None
title: str | None = None
media_type: str | None = None
tmdb_id: int | None = None
release_date: str | None = None
hits: list[MovieHit] = field(default_factory=list)
error: str | None = None
message: str | None = None
def to_dict(self):
"""Convert to dict for agent compatibility."""
result = {"status": self.status}
result: dict = {"status": self.status}
if self.error:
result["error"] = self.error
result["message"] = self.message
else:
if self.imdb_id:
result["imdb_id"] = self.imdb_id
if self.title:
result["title"] = self.title
if self.media_type:
result["media_type"] = self.media_type
if self.tmdb_id:
result["tmdb_id"] = self.tmdb_id
if self.release_date:
result["release_date"] = self.release_date
result["hits"] = [h.to_dict() for h in self.hits]
return result
alfred/application/movies/search_movie.py
+17 -48
View File
@@ -6,71 +6,40 @@ from alfred.infrastructure.api.tmdb import (
TMDBAPIError,
TMDBClient,
TMDBConfigurationError,
TMDBNotFoundError,
)
from .dto import SearchMovieResponse
from .dto import MovieHit, SearchMovieResponse
logger = logging.getLogger(__name__)
class SearchMovieUseCase:
"""
Use case for searching a movie and retrieving its IMDb ID.
"""List movies matching a free-text query via TMDB ``/search/movie``.
This orchestrates the TMDB API client to find movie information.
The use case is a thin orchestrator: it asks the client for hits,
flattens domain VOs into agent-friendly primitives, and wraps
errors. It deliberately does **not** look up ``imdb_id`` —
enrichment is the caller's job (via :meth:`TMDBClient.get_movie_info`
on a chosen ``tmdb_id``).
"""
def __init__(self, tmdb_client: TMDBClient):
"""
Initialize use case.
Args:
tmdb_client: TMDB API client
"""
self.tmdb_client = tmdb_client
def execute(self, media_title: str) -> SearchMovieResponse:
"""
Search for a movie by title.
Args:
media_title: Title of the movie to search for
Returns:
SearchMovieResponse with movie information or error
"""
try:
# Use the TMDB client to search for media
result = self.tmdb_client.search_media(media_title)
results = self.tmdb_client.search_movies(media_title)
# Check if IMDb ID was found # FUCK THIS
if result.imdb_id:
logger.info(f"IMDb ID found for '{media_title}': {result.imdb_id}")
return SearchMovieResponse(
status="ok",
imdb_id=result.imdb_id,
title=result.title,
media_type=result.media_type,
tmdb_id=result.tmdb_id,
release_date=result.release_date,
)
else:
logger.warning(f"No IMDb ID available for '{media_title}'")
return SearchMovieResponse(
status="ok",
title=result.title,
media_type=result.media_type,
tmdb_id=result.tmdb_id,
error="no_imdb_id",
message=f"No IMDb ID available for '{result.title}'",
)
except TMDBNotFoundError as e:
logger.info(f"Media not found: {e}")
return SearchMovieResponse(
status="error", error="not_found", message=str(e)
hits = [
MovieHit(
tmdb_id=r.tmdb_id.value,
title=str(r.title),
release_year=r.release_year.value if r.release_year else None,
)
for r in results
]
logger.info(f"search_movies({media_title!r}) → {len(hits)} hits")
return SearchMovieResponse(status="ok", hits=hits)
except TMDBConfigurationError as e:
logger.error(f"TMDB configuration error: {e}")
alfred/application/tv_shows/__init__.py
+10 -2
View File
@@ -7,7 +7,15 @@ reusing the existing release pipeline (``inspect_release``) rather
than duplicating its parse/probe logic.
"""
from .rescan import rescan_show
from .dto import SearchShowResponse, ShowHit
from .search_show import SearchShowUseCase
from .walker import SeasonFolder, ShowTree, walk_show
__all__ = ["SeasonFolder", "ShowTree", "rescan_show", "walk_show"]
__all__ = [
"SearchShowResponse",
"SearchShowUseCase",
"SeasonFolder",
"ShowHit",
"ShowTree",
"walk_show",
]
alfred/application/tv_shows/dto.py
+39
View File
@@ -0,0 +1,39 @@
"""TV show application DTOs."""
from dataclasses import dataclass, field
@dataclass(frozen=True)
class ShowHit:
"""One TV-show hit, flattened for transport to the agent."""
tmdb_id: int
name: str
first_air_year: int | None = None
def to_dict(self) -> dict:
out: dict = {"tmdb_id": self.tmdb_id, "name": self.name}
if self.first_air_year is not None:
out["first_air_year"] = self.first_air_year
return out
@dataclass
class SearchShowResponse:
"""Response from searching for a TV show."""
status: str
hits: list[ShowHit] = field(default_factory=list)
error: str | None = None
message: str | None = None
def to_dict(self):
result: dict = {"status": self.status}
if self.error:
result["error"] = self.error
result["message"] = self.message
else:
result["hits"] = [h.to_dict() for h in self.hits]
return result
alfred/application/tv_shows/search_show.py
+59
View File
@@ -0,0 +1,59 @@
"""Search TV show use case."""
import logging
from alfred.infrastructure.api.tmdb import (
TMDBAPIError,
TMDBClient,
TMDBConfigurationError,
)
from .dto import SearchShowResponse, ShowHit
logger = logging.getLogger(__name__)
class SearchShowUseCase:
"""List TV shows matching a free-text query via TMDB ``/search/tv``.
Symmetric to :class:`alfred.application.movies.SearchMovieUseCase`:
thin orchestrator, flattens domain VOs into agent-friendly
primitives, no ``imdb_id`` enrichment (caller follows up with
:meth:`TMDBClient.get_tv_show_info` on a chosen ``tmdb_id``).
"""
def __init__(self, tmdb_client: TMDBClient):
self.tmdb_client = tmdb_client
def execute(self, show_title: str) -> SearchShowResponse:
try:
results = self.tmdb_client.search_shows(show_title)
hits = [
ShowHit(
tmdb_id=r.tmdb_id.value,
name=r.name,
first_air_year=r.first_air_year,
)
for r in results
]
logger.info(f"search_shows({show_title!r}) → {len(hits)} hits")
return SearchShowResponse(status="ok", hits=hits)
except TMDBConfigurationError as e:
logger.error(f"TMDB configuration error: {e}")
return SearchShowResponse(
status="error", error="configuration_error", message=str(e)
)
except TMDBAPIError as e:
logger.error(f"TMDB API error: {e}")
return SearchShowResponse(
status="error", error="api_error", message=str(e)
)
except ValueError as e:
logger.error(f"Validation error: {e}")
return SearchShowResponse(
status="error", error="validation_failed", message=str(e)
)
alfred/domain/movies/value_objects.py
+2 -2
View File
@@ -55,9 +55,9 @@ class MovieTitle:
f"Movie title must be a string, got {type(self.value)}"
)
if len(self.value) > 100:
if len(self.value) > 150:
raise ValidationError(
f"Movie title too long: {len(self.value)} characters (max 100)"
f"Movie title too long: {len(self.value)} characters (max 150)"
)
alfred/domain/shared/__init__.py
+2
View File
@@ -1,11 +1,13 @@
"""Shared kernel - Common domain concepts used across subdomains."""
from .exceptions import DomainException, ValidationError
from .file_entry import FileEntry
from .value_objects import FilePath, FileSize, ImdbId, Language
__all__ = [
"DomainException",
"ValidationError",
"FileEntry",
"ImdbId",
"FilePath",
"FileSize",
alfred/domain/shared/file_entry.py
+39
View File
@@ -0,0 +1,39 @@
"""FileEntry — frozen snapshot of one filesystem entry.
Produced by a ``FilesystemScanner`` adapter and consumed by the domain.
The domain never calls ``Path.iterdir``, ``Path.is_file``, ``Path.stat``
or ``open()`` directly; it reasons from these snapshots only. One scan =
one I/O round-trip; no callbacks back to disk.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class FileEntry:
"""Frozen snapshot of one filesystem entry, taken at scan time.
The entry carries enough metadata for the domain to classify and order
files without re-querying the OS. ``size`` is expressed in bytes and is
``None`` for directories and for files whose size could not be read.
"""
path: Path
is_file: bool
is_dir: bool
size: int | None
@property
def name(self) -> str:
return self.path.name
@property
def stem(self) -> str:
return self.path.stem
@property
def suffix(self) -> str:
return self.path.suffix
alfred/domain/shared/ports/__init__.py
+1 -2
View File
@@ -6,12 +6,11 @@ injection and calls it. Tests can pass in-memory fakes that satisfy the
Protocol without going through real I/O.
"""
from .filesystem_scanner import FileEntry, FilesystemScanner
from .filesystem_scanner import FilesystemScanner
from .language_repository import LanguageRepository
from .media_prober import MediaProber, SubtitleStreamInfo
__all__ = [
"FileEntry",
"FilesystemScanner",
"LanguageRepository",
"MediaProber",
alfred/domain/shared/ports/filesystem_scanner.py
+1 -27
View File
@@ -7,36 +7,10 @@ reasons from there. One scan = one I/O round-trip; no callbacks back to disk.
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
@dataclass(frozen=True)
class FileEntry:
"""Frozen snapshot of one filesystem entry, taken at scan time.
The entry carries enough metadata for the domain to classify and order
files without re-querying the OS. ``size_kb`` is ``None`` for directories
and for files whose size could not be read.
"""
path: Path
is_file: bool
is_dir: bool
size_kb: float | None
@property
def name(self) -> str:
return self.path.name
@property
def stem(self) -> str:
return self.path.stem
@property
def suffix(self) -> str:
return self.path.suffix
from alfred.domain.shared.file_entry import FileEntry
class FilesystemScanner(Protocol):
alfred/domain/subtitles/services/identifier.py
+1 -1
View File
@@ -259,7 +259,7 @@ class SubtitleIdentifier:
subtitle_type=subtitle_type,
is_embedded=False,
file_path=entry.path,
file_size_kb=entry.size_kb,
file_size_kb=(entry.size / 1024) if entry.size is not None else None,
entry_count=entry_count,
confidence=confidence,
raw_tokens=tokens,
alfred/domain/tv_shows/builders.py
+5 -5
View File
@@ -11,7 +11,7 @@ Typical usage during a TMDB hydration::
tmdb_id=TmdbId(1396),
title="Breaking Bad",
imdb_id=ImdbId("tt0903747"),
status="Ended",
status=ShowStatus.ENDED,
)
builder.season_builder(1).set_episode_count(7).add_episode(Episode(
season_number=SeasonNumber(1),
@@ -45,7 +45,7 @@ from __future__ import annotations
from ..shared.value_objects import ImdbId, TmdbId
from .entities import Episode, Season, TVShow
from .value_objects import EpisodeNumber, SeasonNumber
from .value_objects import EpisodeNumber, SeasonNumber, ShowStatus
# ════════════════════════════════════════════════════════════════════════════
# SeasonBuilder
@@ -139,7 +139,7 @@ class TVShowBuilder:
tmdb_id: TmdbId,
title: str,
imdb_id: ImdbId | None = None,
status: str = "unknown",
status: ShowStatus = ShowStatus.UNKNOWN,
) -> None:
if not isinstance(tmdb_id, TmdbId):
raise ValueError(
@@ -152,7 +152,7 @@ class TVShowBuilder:
self._tmdb_id: TmdbId = tmdb_id
self._title: str = title
self._imdb_id: ImdbId | None = imdb_id
self._status: str = status
self._status: ShowStatus = status
self._season_builders: dict[SeasonNumber, SeasonBuilder] = {}
@classmethod
@@ -184,7 +184,7 @@ class TVShowBuilder:
self._imdb_id = imdb_id
return self
def set_status(self, status: str) -> TVShowBuilder:
def set_status(self, status: ShowStatus) -> TVShowBuilder:
self._status = status
return self
alfred/domain/tv_shows/entities.py
+5 -7
View File
@@ -42,7 +42,7 @@ from ..shared.value_objects import (
TmdbId,
to_dot_folder_name,
)
from .value_objects import EpisodeNumber, SeasonNumber
from .value_objects import EpisodeNumber, SeasonNumber, ShowStatus
# ════════════════════════════════════════════════════════════════════════════
# Episode
@@ -208,17 +208,15 @@ class TVShow:
Identity is carried by ``tmdb_id`` (primary key, required) and
``imdb_id`` (optional secondary anchor — some TMDB items lack one).
:attr:`status` mirrors the raw TMDB status string verbatim
(``"Returning Series"`` / ``"Ended"`` / ``"Canceled"`` …). No
taxonomy of our own — callers that need a normalized view can map
it themselves. The ``"unknown"`` default matches the auto-heal
placeholder used by the v2 library index.
:attr:`status` is a :class:`ShowStatus` — the TMDB-native lifecycle
enum. :attr:`ShowStatus.UNKNOWN` is the default placeholder used by
the v2 library index when no status has been synced yet.
"""
tmdb_id: TmdbId
title: str
imdb_id: ImdbId | None = None
status: str = "unknown"
status: ShowStatus = ShowStatus.UNKNOWN
seasons: tuple[Season, ...] = ()
def __post_init__(self) -> None:
alfred/domain/tv_shows/value_objects.py
+42
View File
@@ -2,11 +2,53 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from ..shared.exceptions import ValidationError
logger = logging.getLogger(__name__)
class ShowStatus(Enum):
"""Lifecycle status of a TV show.
Values mirror TMDB's ``status`` strings verbatim so the boundary
code (``parse_tv_show_info``) maps one-to-one without an opaque
translation table buried elsewhere. If TMDB ships a new status,
:meth:`from_tmdb` falls back to :attr:`UNKNOWN` and emits a
warning — sync keeps working, the gap is visible in logs.
"""
RETURNING_SERIES = "Returning Series"
PLANNED = "Planned"
IN_PRODUCTION = "In Production"
ENDED = "Ended"
CANCELED = "Canceled"
PILOT = "Pilot"
UNKNOWN = "Unknown"
@classmethod
def from_tmdb(cls, raw: str) -> ShowStatus:
"""Map a TMDB ``status`` string to a :class:`ShowStatus`.
Unknown strings fall back to :attr:`UNKNOWN` with a logged
warning — TMDB occasionally adds new lifecycle states and we
prefer to keep syncing rather than break the whole pipeline.
"""
for member in cls:
if member.value == raw:
return member
logger.warning(
f"ShowStatus: unrecognized TMDB status {raw!r}, "
f"falling back to ShowStatus.UNKNOWN"
)
return cls.UNKNOWN
def __str__(self) -> str:
return self.value
class SeasonMode(Enum):
"""
alfred/infrastructure/api/tmdb/__init__.py
+19 -6
View File
@@ -1,7 +1,15 @@
"""TMDB API client."""
from .client import TMDBClient
from .dto import ExternalIds, MediaResult
from .dto import (
TmdbMovieInfo,
TmdbMovieSearchResult,
TmdbSeasonInfo,
TmdbShowInfo,
TmdbShowSearchResult,
parse_movie_info,
parse_tv_show_info,
)
from .exceptions import (
TMDBAPIError,
TMDBConfigurationError,
@@ -13,12 +21,17 @@ from .exceptions import (
tmdb_client = TMDBClient()
__all__ = [
"TMDBClient",
"MediaResult",
"ExternalIds",
"TMDBError",
"TMDBConfigurationError",
"TMDBAPIError",
"TMDBClient",
"TMDBConfigurationError",
"TMDBError",
"TMDBNotFoundError",
"TmdbMovieInfo",
"TmdbMovieSearchResult",
"TmdbSeasonInfo",
"TmdbShowInfo",
"TmdbShowSearchResult",
"parse_movie_info",
"parse_tv_show_info",
"tmdb_client",
]
alfred/infrastructure/api/tmdb/client.py
+84 -99
View File
@@ -1,6 +1,7 @@
"""TMDB (The Movie Database) API client."""
import logging
from datetime import date
from typing import Any
import requests
@@ -8,10 +9,14 @@ from requests.exceptions import HTTPError, RequestException, Timeout
from alfred.settings import Settings, settings
from alfred.domain.movies.value_objects import MovieTitle, ReleaseYear
from alfred.domain.shared.value_objects import TmdbId
from .dto import (
MediaResult,
TmdbMovieInfo,
TmdbMovieSearchResult,
TmdbShowInfo,
TmdbShowSearchResult,
parse_movie_info,
parse_tv_show_info,
)
@@ -24,6 +29,22 @@ from .exceptions import (
logger = logging.getLogger(__name__)
def _parse_year(raw: Any) -> int | None:
"""Extract the year from a TMDB ISO date (``YYYY-MM-DD``).
Returns ``None`` if ``raw`` is missing, empty, or not a valid
ISO date — TMDB occasionally returns ``""`` for unscheduled
releases. Symmetric to the parsing done in
:mod:`alfred.infrastructure.api.tmdb.dto`.
"""
if not isinstance(raw, str) or not raw:
return None
try:
return date.fromisoformat(raw).year
except ValueError:
return None
class TMDBClient:
"""
Client for interacting with The Movie Database (TMDB) API.
@@ -33,9 +54,9 @@ class TMDBClient:
Example:
>>> client = TMDBClient()
>>> result = client.search_media("Inception")
>>> print(result.imdb_id)
'tt1375666'
>>> results = client.search_movies("Inception")
>>> print(results[0].title)
MovieTitle('Inception')
"""
def __init__(
@@ -123,34 +144,79 @@ class TMDBClient:
logger.error(f"TMDB API request failed: {e}")
raise TMDBAPIError(f"Failed to connect to TMDB API: {e}") from e
def search_multi(self, query: str) -> list[dict[str, Any]]:
"""
Search for movies and TV shows.
def search_movies(self, query: str) -> list[TmdbMovieSearchResult]:
"""Search the TMDB ``/search/movie`` endpoint.
Args:
query: Search query (movie or TV show title)
query: free-text title query.
Returns:
List of search results
All hits as :class:`TmdbMovieSearchResult` (empty list if
none — no exception). Identity is wrapped in domain VOs
(``TmdbId``, ``MovieTitle``, ``ReleaseYear``). No
``external_ids`` enrichment: callers that need ``imdb_id``
should follow up with :meth:`get_movie_info` on the chosen
hit's ``tmdb_id``.
Raises:
TMDBAPIError: If request fails
TMDBNotFoundError: If no results found
TMDBAPIError: if the HTTP call fails.
ValueError: if ``query`` is empty / too long.
"""
if not query or not isinstance(query, str):
raise ValueError("Query must be a non-empty string")
if len(query) > 500:
raise ValueError("Query is too long (max 500 characters)")
data = self._make_request("/search/multi", {"query": query})
data = self._make_request("/search/movie", {"query": query})
raw_results = data.get("results", []) or []
logger.info(f"search_movies({query!r}) → {len(raw_results)} hits")
return [self._parse_movie_hit(r) for r in raw_results]
results = data.get("results", [])
if not results:
raise TMDBNotFoundError(f"No results found for '{query}'")
def search_shows(self, query: str) -> list[TmdbShowSearchResult]:
"""Search the TMDB ``/search/tv`` endpoint. Symmetric to
:meth:`search_movies`.
logger.info(f"Found {len(results)} results for '{query}'")
return results
Args:
query: free-text title query.
Returns:
All hits as :class:`TmdbShowSearchResult` (empty list if
none). ``tmdb_id`` is wrapped as :class:`TmdbId`; ``name``
stays plain ``str`` (no filename-safe VO for shows).
``first_air_year`` is parsed from ``first_air_date``.
Raises:
TMDBAPIError: if the HTTP call fails.
ValueError: if ``query`` is empty / too long.
"""
if not query or not isinstance(query, str):
raise ValueError("Query must be a non-empty string")
if len(query) > 500:
raise ValueError("Query is too long (max 500 characters)")
data = self._make_request("/search/tv", {"query": query})
raw_results = data.get("results", []) or []
logger.info(f"search_shows({query!r}) → {len(raw_results)} hits")
return [self._parse_show_hit(r) for r in raw_results]
@staticmethod
def _parse_movie_hit(raw: dict[str, Any]) -> TmdbMovieSearchResult:
title_raw = raw.get("title") or raw.get("original_title") or "Unknown"
year = _parse_year(raw.get("release_date"))
return TmdbMovieSearchResult(
tmdb_id=TmdbId(raw["id"]),
title=MovieTitle(title_raw),
release_year=ReleaseYear(year) if year is not None else None,
)
@staticmethod
def _parse_show_hit(raw: dict[str, Any]) -> TmdbShowSearchResult:
name_raw = raw.get("name") or raw.get("original_name") or "Unknown"
return TmdbShowSearchResult(
tmdb_id=TmdbId(raw["id"]),
name=name_raw,
first_air_year=_parse_year(raw.get("first_air_date")),
)
def get_external_ids(self, media_type: str, tmdb_id: int) -> dict[str, Any]:
"""
@@ -174,87 +240,6 @@ class TMDBClient:
endpoint = f"/{media_type}/{tmdb_id}/external_ids"
return self._make_request(endpoint)
def search_media(self, title: str) -> MediaResult:
"""
Search for a media item and return detailed information including IMDb ID.
This is a convenience method that combines search and external ID lookup.
Args:
title: Title of the movie or TV show
Returns:
MediaResult with all available information
Raises:
TMDBAPIError: If request fails
TMDBNotFoundError: If media not found
"""
# Search for media
results = self.search_multi(title)
# Get the first (most relevant) result
top_result = results[0]
# Validate result structure
if "id" not in top_result or "media_type" not in top_result:
raise TMDBAPIError("Invalid TMDB response structure")
media_type = top_result["media_type"]
# Skip if not movie or TV show
if media_type not in ("movie", "tv"):
logger.warning(f"Skipping result of type: {media_type}")
if len(results) > 1:
# Try next result
return self._parse_result(results[1])
raise TMDBNotFoundError(f"No movie or TV show found for '{title}'")
return self._parse_result(top_result)
def _parse_result(self, result: dict[str, Any]) -> MediaResult:
"""
Parse a TMDB result into a MediaResult object.
Args:
result: Raw TMDB result dict
Returns:
MediaResult object
"""
tmdb_id = result["id"]
media_type = result["media_type"]
title = result.get("title") or result.get("name", "Unknown")
# Get external IDs (including IMDb)
try:
external_ids = self.get_external_ids(media_type, tmdb_id)
imdb_id = external_ids.get("imdb_id")
except TMDBAPIError as e:
logger.warning(f"Failed to get external IDs: {e}")
imdb_id = None
# Extract other useful information
overview = result.get("overview")
release_date = result.get("release_date") or result.get("first_air_date")
poster_path = result.get("poster_path")
vote_average = result.get("vote_average")
logger.info(
f"Found: {title} (Type: {media_type}, TMDB ID: {tmdb_id}, IMDb: {imdb_id})"
)
return MediaResult(
tmdb_id=tmdb_id,
title=title,
media_type=media_type,
imdb_id=imdb_id,
overview=overview,
release_date=release_date,
poster_path=poster_path,
vote_average=vote_average,
)
def get_movie_details(self, movie_id: int) -> dict[str, Any]:
"""
Get detailed information about a movie.
alfred/infrastructure/api/tmdb/dto.py
+81 -50
View File
@@ -6,30 +6,39 @@ from dataclasses import dataclass, field
from datetime import date
from typing import Any
@dataclass
class MediaResult:
"""Represents a media search result from TMDB."""
tmdb_id: int
title: str
media_type: str # 'movie' or 'tv'
imdb_id: str | None = None
overview: str | None = None
release_date: str | None = None
poster_path: str | None = None
vote_average: float | None = None
from alfred.domain.movies.value_objects import MovieTitle, ReleaseYear
from alfred.domain.shared.value_objects import ImdbId, TmdbId
from alfred.domain.tv_shows.value_objects import ShowStatus
@dataclass
class ExternalIds:
"""External IDs for a media item."""
@dataclass(frozen=True)
class TmdbMovieSearchResult:
"""One movie hit from the TMDB ``/search/movie`` endpoint.
imdb_id: str | None = None
tvdb_id: int | None = None
facebook_id: str | None = None
instagram_id: str | None = None
twitter_id: str | None = None
Carries the minimal identity + display facts needed by the search
use case. No external-ids enrichment — that requires a separate
HTTP call and is out of scope for search results. Callers that
need ``imdb_id`` for a specific hit should subsequently call
:meth:`TMDBClient.get_movie_info` with the ``tmdb_id``.
"""
tmdb_id: TmdbId
title: MovieTitle
release_year: ReleaseYear | None = None
@dataclass(frozen=True)
class TmdbShowSearchResult:
"""One TV-show hit from the TMDB ``/search/tv`` endpoint.
Symmetric to :class:`TmdbMovieSearchResult`. Uses ``name`` (TMDB's
TV equivalent of ``title``) and ``first_air_year`` (parsed from
``first_air_date``). No external-ids enrichment.
"""
tmdb_id: TmdbId
name: str
first_air_year: int | None = None
# ────────────────────────────────────────────────────────────────────────────
@@ -58,12 +67,19 @@ class TmdbShowInfo:
Populated by :meth:`TMDBClient.get_tv_show_info`. Carries only the
fields the v2 library index needs to cache; richer details remain
on-demand via the raw client methods.
Identity fields are typed as value objects from the domain shared
kernel (``TmdbId``, ``ImdbId``) so validation happens at the
boundary with TMDB. ``status`` is a :class:`ShowStatus` — TMDB's
raw string is mapped via :meth:`ShowStatus.from_tmdb` at parse
time. ``name`` stays plain ``str``: shows don't have a filename-
safe wrapper analogous to :class:`MovieTitle`.
"""
tmdb_id: int
imdb_id: str | None
tmdb_id: TmdbId
imdb_id: ImdbId | None
name: str
status: str
status: ShowStatus
seasons: tuple[TmdbSeasonInfo, ...] = field(default_factory=tuple)
@@ -93,33 +109,37 @@ def parse_tv_show_info(
"""
ref = today or date.today()
tmdb_id = details.get("id")
if not isinstance(tmdb_id, int):
tmdb_id_raw = details.get("id")
if not isinstance(tmdb_id_raw, int):
raise ValueError(
f"TMDB show payload missing/invalid 'id': {tmdb_id!r}"
f"TMDB show payload missing/invalid 'id': {tmdb_id_raw!r}"
)
name = details.get("name")
if not isinstance(name, str) or not name:
raise ValueError(
f"TMDB show payload missing/invalid 'name': {name!r}"
)
status = details.get("status")
if not isinstance(status, str) or not status:
status_raw = details.get("status")
if not isinstance(status_raw, str) or not status_raw:
raise ValueError(
f"TMDB show payload missing/invalid 'status': {status!r}"
f"TMDB show payload missing/invalid 'status': {status_raw!r}"
)
imdb_id_raw = external_ids.get("imdb_id")
imdb_id = imdb_id_raw if isinstance(imdb_id_raw, str) and imdb_id_raw else None
imdb_id = (
ImdbId(imdb_id_raw)
if isinstance(imdb_id_raw, str) and imdb_id_raw
else None
)
seasons_raw = details.get("seasons", []) or []
seasons = tuple(_parse_season(s, ref) for s in seasons_raw)
return TmdbShowInfo(
tmdb_id=tmdb_id,
tmdb_id=TmdbId(tmdb_id_raw),
imdb_id=imdb_id,
name=name,
status=status,
status=ShowStatus.from_tmdb(status_raw),
seasons=seasons,
)
@@ -172,15 +192,22 @@ class TmdbMovieInfo:
remain on-demand via the raw client methods. Symmetric to
:class:`TmdbShowInfo`.
Identity fields are typed as value objects from the domain shared
kernel (``TmdbId``, ``ImdbId``) so validation happens at the
boundary with TMDB, not three layers later. Likewise ``title``
and ``release_year`` are wrapped as VOs (``MovieTitle``,
``ReleaseYear``) — purely structural wrappers that just enforce
type and (for title) a 150-char filename-safe cap.
``release_year`` is parsed from TMDB's ``release_date`` (the
first 4 chars). It is ``None`` when TMDB has no release date
(very old or future-dated titles).
"""
tmdb_id: int
imdb_id: str | None
title: str
release_year: int | None
tmdb_id: TmdbId
imdb_id: ImdbId | None
title: MovieTitle
release_year: ReleaseYear | None
def parse_movie_info(
@@ -203,27 +230,31 @@ def parse_movie_info(
ValueError: if a required field (``id``, ``title``) is missing
from ``details``.
"""
tmdb_id = details.get("id")
if not isinstance(tmdb_id, int):
tmdb_id_raw = details.get("id")
if not isinstance(tmdb_id_raw, int):
raise ValueError(
f"TMDB movie payload missing/invalid 'id': {tmdb_id!r}"
f"TMDB movie payload missing/invalid 'id': {tmdb_id_raw!r}"
)
title = details.get("title")
if not isinstance(title, str) or not title:
title_raw = details.get("title")
if not isinstance(title_raw, str) or not title_raw:
raise ValueError(
f"TMDB movie payload missing/invalid 'title': {title!r}"
f"TMDB movie payload missing/invalid 'title': {title_raw!r}"
)
imdb_id_raw = external_ids.get("imdb_id")
imdb_id = imdb_id_raw if isinstance(imdb_id_raw, str) and imdb_id_raw else None
imdb_id = (
ImdbId(imdb_id_raw)
if isinstance(imdb_id_raw, str) and imdb_id_raw
else None
)
release_year = _parse_release_year(details.get("release_date"))
release_year_raw = _parse_release_year(details.get("release_date"))
return TmdbMovieInfo(
tmdb_id=tmdb_id,
tmdb_id=TmdbId(tmdb_id_raw),
imdb_id=imdb_id,
title=title,
release_year=release_year,
title=MovieTitle(title_raw),
release_year=ReleaseYear(release_year_raw) if release_year_raw is not None else None,
)
@@ -232,9 +263,9 @@ def _parse_release_year(release_date_raw: Any) -> int | None:
Returns ``None`` for missing, empty, or unparseable values.
"""
if not isinstance(release_date_raw, str) or len(release_date_raw) < 4:
if not isinstance(release_date_raw, str) or not release_date_raw:
return None
try:
return int(release_date_raw[:4])
return date.fromisoformat(release_date_raw).year
except ValueError:
return None
alfred/infrastructure/filesystem/scanner.py
+5 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import logging
from pathlib import Path
from alfred.domain.shared.ports import FileEntry
from alfred.domain.shared import FileEntry
logger = logging.getLogger(__name__)
@@ -56,11 +56,11 @@ class PathlibFilesystemScanner:
if not (is_file or is_dir):
return None
size_kb: float | None = None
size: int | None = None
if is_file:
try:
size_kb = path.stat().st_size / 1024
size = path.stat().st_size
except OSError:
size_kb = None
size = None
return FileEntry(path=path, is_file=is_file, is_dir=is_dir, size_kb=size_kb)
return FileEntry(path=path, is_file=is_file, is_dir=is_dir, size=size)
alfred/infrastructure/persistence/dot_alfred/v2/bridge.py
+7 -8
View File
@@ -217,10 +217,10 @@ def show_index_entry_from(
for s in info.seasons
)
return ShowIndexEntry(
tmdb_id=info.tmdb_id,
imdb_id=info.imdb_id,
tmdb_id=info.tmdb_id.value,
imdb_id=str(info.imdb_id) if info.imdb_id is not None else None,
name=info.name,
status=info.status,
status=info.status.value,
metadata=ShowIndexMetadata(path=path, fetched_at=fetched_at),
seasons=seasons,
)
@@ -236,11 +236,10 @@ def movie_index_entry_from(
) -> MovieIndexEntry:
"""Project a movie release + identity facts into one library-index entry.
Movies don't have a ``TmdbMovieInfo`` DTO yet (no per-movie TMDB
cache surface defined in Phase 2), so identity facts are passed
explicitly by the caller. The release supplies ``tmdb_id`` /
``imdb_id``; ``name`` and ``release_year`` come from the caller's
TMDB lookup (or a future ``TmdbMovieInfo`` DTO when one ships).
Identity facts are passed explicitly by the caller (the release
supplies ``tmdb_id`` / ``imdb_id``; ``name`` and ``release_year``
typically come from a ``TmdbMovieInfo`` lookup that the caller
unwraps to primitives before passing in).
"""
return MovieIndexEntry(
tmdb_id=release.tmdb_id.value,
alfred/infrastructure/persistence/dot_alfred/v2/repository.py
+2 -1
View File
@@ -40,6 +40,7 @@ from pydantic import ValidationError
from ....api.tmdb.dto import TmdbShowInfo
from .....domain.releases.entities import MovieRelease, SeriesRelease
from .....domain.shared.value_objects import ImdbId, TmdbId
from .....domain.tv_shows.value_objects import ShowStatus
from .bridge import (
movie_index_entry_from,
movie_release_from_sidecar,
@@ -455,7 +456,7 @@ class DotAlfredTVShowLibraryIndex:
tmdb_id=release.tmdb_id.value,
imdb_id=str(release.imdb_id) if release.imdb_id else None,
name=folder, # placeholder until TMDB sync supplies the real name
status="unknown", # placeholder until TMDB sync supplies status
status=ShowStatus.UNKNOWN.value, # placeholder until TMDB sync supplies status
metadata={
"path": folder,
"fetched_at": now,
tests/application/test_search_movie.py
+47 -57
View File
@@ -1,15 +1,16 @@
"""Tests for ``alfred.application.movies.search_movie.SearchMovieUseCase``.
The use case wraps ``TMDBClient.search_media`` and converts results / errors
into a ``SearchMovieResponse`` envelope (status="ok"|"error").
The use case wraps :meth:`TMDBClient.search_movies` and flattens each
hit's domain VOs into agent-friendly primitives wrapped in a
:class:`SearchMovieResponse` envelope (status="ok"|"error").
Coverage:
- ``TestSuccess`` — full MediaResult with imdb_id → ok+imdb_id; missing
imdb_id → ok+no_imdb_id; TV media_type preserved.
- ``TestErrorTranslation`` — ``TMDBNotFoundError`` → not_found,
``TMDBConfigurationError`` → configuration_error,
``TMDBAPIError`` → api_error, ``ValueError`` → validation_failed.
- ``TestSuccess`` — list of hits flattened, year present/absent,
empty list still ``status="ok"``.
- ``TestErrorTranslation`` — ``TMDBConfigurationError`` →
configuration_error, ``TMDBAPIError`` → api_error, ``ValueError``
→ validation_failed.
- ``TestPassThrough`` — query is forwarded to the client unchanged.
TMDBClient is fully mocked — no real HTTP.
@@ -22,11 +23,12 @@ from unittest.mock import MagicMock
import pytest
from alfred.application.movies.search_movie import SearchMovieUseCase
from alfred.infrastructure.api.tmdb.dto import MediaResult
from alfred.domain.movies.value_objects import MovieTitle, ReleaseYear
from alfred.domain.shared.value_objects import TmdbId
from alfred.infrastructure.api.tmdb.dto import TmdbMovieSearchResult
from alfred.infrastructure.api.tmdb.exceptions import (
TMDBAPIError,
TMDBConfigurationError,
TMDBNotFoundError,
)
@@ -40,19 +42,14 @@ def use_case(client):
return SearchMovieUseCase(client)
def _result(**kw) -> MediaResult:
def _hit(**kw) -> TmdbMovieSearchResult:
defaults = dict(
tmdb_id=1,
title="Inception",
media_type="movie",
imdb_id="tt1375666",
overview="o",
release_date="2010-07-15",
poster_path="/x.jpg",
vote_average=8.4,
tmdb_id=TmdbId(27205),
title=MovieTitle("Inception"),
release_year=ReleaseYear(2010),
)
defaults.update(kw)
return MediaResult(**defaults)
return TmdbMovieSearchResult(**defaults)
# --------------------------------------------------------------------------- #
@@ -61,36 +58,36 @@ def _result(**kw) -> MediaResult:
class TestSuccess:
def test_full_result_returns_ok_with_imdb_id(self, client, use_case):
client.search_media.return_value = _result()
def test_single_hit_is_flattened(self, client, use_case):
client.search_movies.return_value = [_hit()]
r = use_case.execute("Inception")
assert r.status == "ok"
assert r.imdb_id == "tt1375666"
assert r.title == "Inception"
assert r.media_type == "movie"
assert r.tmdb_id == 1
assert r.vote_average == 8.4
assert len(r.hits) == 1
h = r.hits[0]
assert h.tmdb_id == 27205
assert h.title == "Inception"
assert h.release_year == 2010
def test_multiple_hits_preserve_order(self, client, use_case):
client.search_movies.return_value = [
_hit(),
_hit(tmdb_id=TmdbId(42), title=MovieTitle("Inception 2")),
]
r = use_case.execute("Inception")
assert [h.tmdb_id for h in r.hits] == [27205, 42]
def test_hit_without_release_year(self, client, use_case):
client.search_movies.return_value = [_hit(release_year=None)]
r = use_case.execute("Inception")
assert r.hits[0].release_year is None
def test_empty_results_returns_ok_with_no_hits(self, client, use_case):
client.search_movies.return_value = []
r = use_case.execute("nothing")
assert r.status == "ok"
assert r.hits == []
assert r.error is None
def test_tv_result(self, client, use_case):
client.search_media.return_value = _result(
media_type="tv", title="Breaking Bad", imdb_id="tt0903747"
)
r = use_case.execute("Breaking Bad")
assert r.status == "ok"
assert r.media_type == "tv"
assert r.imdb_id == "tt0903747"
def test_missing_imdb_id_returns_ok_with_no_imdb_id_error(self, client, use_case):
client.search_media.return_value = _result(imdb_id=None)
r = use_case.execute("Inception")
assert r.status == "ok"
assert r.error == "no_imdb_id"
assert r.message is not None
assert "Inception" in r.message
assert r.imdb_id is None
assert r.title == "Inception"
# --------------------------------------------------------------------------- #
# Error translation #
@@ -98,28 +95,21 @@ class TestSuccess:
class TestErrorTranslation:
def test_not_found(self, client, use_case):
client.search_media.side_effect = TMDBNotFoundError("no match")
r = use_case.execute("ghost")
assert r.status == "error"
assert r.error == "not_found"
assert "no match" in r.message
def test_configuration_error(self, client, use_case):
client.search_media.side_effect = TMDBConfigurationError("missing key")
client.search_movies.side_effect = TMDBConfigurationError("missing key")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "configuration_error"
def test_api_error(self, client, use_case):
client.search_media.side_effect = TMDBAPIError("500 oops")
client.search_movies.side_effect = TMDBAPIError("500 oops")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "api_error"
assert "500" in r.message
def test_validation_error(self, client, use_case):
client.search_media.side_effect = ValueError("query too long")
client.search_movies.side_effect = ValueError("query too long")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "validation_failed"
@@ -133,6 +123,6 @@ class TestErrorTranslation:
class TestPassThrough:
def test_query_forwarded_verbatim(self, client, use_case):
client.search_media.return_value = _result()
client.search_movies.return_value = []
use_case.execute("Inception")
client.search_media.assert_called_once_with("Inception")
client.search_movies.assert_called_once_with("Inception")
tests/application/test_search_show.py
+101
View File
@@ -0,0 +1,101 @@
"""Tests for ``alfred.application.tv_shows.search_show.SearchShowUseCase``.
Symmetric to ``test_search_movie.py``. The use case wraps
:meth:`TMDBClient.search_shows` and flattens each hit into a
:class:`ShowHit` wrapped in a :class:`SearchShowResponse` envelope.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from alfred.application.tv_shows.search_show import SearchShowUseCase
from alfred.domain.shared.value_objects import TmdbId
from alfred.infrastructure.api.tmdb.dto import TmdbShowSearchResult
from alfred.infrastructure.api.tmdb.exceptions import (
TMDBAPIError,
TMDBConfigurationError,
)
@pytest.fixture
def client():
return MagicMock()
@pytest.fixture
def use_case(client):
return SearchShowUseCase(client)
def _hit(**kw) -> TmdbShowSearchResult:
defaults = dict(
tmdb_id=TmdbId(84958),
name="Foundation",
first_air_year=2021,
)
defaults.update(kw)
return TmdbShowSearchResult(**defaults)
class TestSuccess:
def test_single_hit_is_flattened(self, client, use_case):
client.search_shows.return_value = [_hit()]
r = use_case.execute("Foundation")
assert r.status == "ok"
assert len(r.hits) == 1
h = r.hits[0]
assert h.tmdb_id == 84958
assert h.name == "Foundation"
assert h.first_air_year == 2021
def test_multiple_hits_preserve_order(self, client, use_case):
client.search_shows.return_value = [
_hit(),
_hit(tmdb_id=TmdbId(42), name="Fallout"),
]
r = use_case.execute("Foundation")
assert [h.tmdb_id for h in r.hits] == [84958, 42]
def test_hit_without_first_air_year(self, client, use_case):
client.search_shows.return_value = [_hit(first_air_year=None)]
r = use_case.execute("Foundation")
assert r.hits[0].first_air_year is None
def test_empty_results_returns_ok_with_no_hits(self, client, use_case):
client.search_shows.return_value = []
r = use_case.execute("nothing")
assert r.status == "ok"
assert r.hits == []
assert r.error is None
class TestErrorTranslation:
def test_configuration_error(self, client, use_case):
client.search_shows.side_effect = TMDBConfigurationError("missing key")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "configuration_error"
def test_api_error(self, client, use_case):
client.search_shows.side_effect = TMDBAPIError("500 oops")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "api_error"
assert "500" in r.message
def test_validation_error(self, client, use_case):
client.search_shows.side_effect = ValueError("query too long")
r = use_case.execute("x")
assert r.status == "error"
assert r.error == "validation_failed"
assert "too long" in r.message
class TestPassThrough:
def test_query_forwarded_verbatim(self, client, use_case):
client.search_shows.return_value = []
use_case.execute("Foundation")
client.search_shows.assert_called_once_with("Foundation")
tests/domain/test_subtitle_identifier.py
+2 -2
View File
@@ -22,7 +22,7 @@ from unittest.mock import patch
import pytest
from alfred.domain.shared.ports import FileEntry
from alfred.domain.shared import FileEntry
from alfred.domain.subtitles.entities import SubtitleScanResult
from alfred.domain.subtitles.services.identifier import (
SubtitleIdentifier,
@@ -48,7 +48,7 @@ def _file_entry(path) -> FileEntry:
path=path,
is_file=path.is_file(),
is_dir=path.is_dir(),
size_kb=(path.stat().st_size / 1024) if path.is_file() else None,
size=path.stat().st_size if path.is_file() else None,
)
tests/infrastructure/api/test_tmdb_client.py
+103 -118
View File
@@ -6,13 +6,12 @@ Exercises the public surface without any real HTTP traffic:
enforcement of the ``api_key``/``base_url`` invariants.
- ``TestMakeRequest`` — error translation for timeouts, HTTP 401/404/5xx,
and generic ``RequestException``.
- ``TestSearchMulti`` — query validation, success path, empty-results →
``TMDBNotFoundError``.
- ``TestSearchMovies`` / ``TestSearchShows`` — query validation, success
path (VO-wrapped hits), empty results yield empty list (no exception).
- ``TestGetExternalIds`` — ``media_type`` whitelist enforcement.
- ``TestSearchMedia`` — happy path (movie/tv), media_type fallthrough to
the next result, structural-validation error, and the case where
external-ID resolution fails but the search still succeeds.
- ``TestDetailsEndpoints`` — ``get_movie_details`` / ``get_tv_details``.
- ``TestGetTvShowInfo`` / ``TestGetMovieInfo`` — composite info getters
aggregating details + external_ids into VO-typed DTOs.
- ``TestIsConfigured`` — reports ``True`` only when both api_key & url set.
All HTTP is mocked at ``alfred.infrastructure.api.tmdb.client.requests``.
@@ -25,8 +24,10 @@ from unittest.mock import MagicMock, patch
import pytest
from requests.exceptions import HTTPError, RequestException, Timeout
from alfred.domain.movies.value_objects import MovieTitle, ReleaseYear
from alfred.domain.shared.value_objects import ImdbId, TmdbId
from alfred.domain.tv_shows.value_objects import ShowStatus
from alfred.infrastructure.api.tmdb.client import TMDBClient
from alfred.infrastructure.api.tmdb.dto import MediaResult
from alfred.infrastructure.api.tmdb.exceptions import (
TMDBAPIError,
TMDBConfigurationError,
@@ -85,7 +86,6 @@ class TestInit:
TMDBClient(api_key="", config=cfg)
def test_missing_base_url_raises(self):
# Pass api_key but force empty base_url. Need a config with empty URL too.
from alfred.settings import Settings
cfg = Settings(tmdb_api_key="fake", tmdb_base_url="")
@@ -140,34 +140,109 @@ class TestMakeRequest:
# --------------------------------------------------------------------------- #
# search_multi #
# search_movies #
# --------------------------------------------------------------------------- #
class TestSearchMulti:
class TestSearchMovies:
@pytest.mark.parametrize("bad", ["", None, 123])
def test_invalid_query_raises_value_error(self, client, bad):
with pytest.raises(ValueError):
client.search_multi(bad)
client.search_movies(bad)
def test_query_too_long(self, client):
with pytest.raises(ValueError, match="too long"):
client.search_multi("a" * 501)
client.search_movies("a" * 501)
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_success(self, mock_get, client):
def test_success_wraps_vos(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"id": 1, "media_type": "movie"}]}
{
"results": [
{
"id": 27205,
"title": "Inception",
"release_date": "2010-07-15",
}
]
}
)
results = client.search_multi("Inception")
results = client.search_movies("Inception")
assert len(results) == 1
assert results[0]["id"] == 1
hit = results[0]
assert hit.tmdb_id == TmdbId(27205)
assert hit.title == MovieTitle("Inception")
assert hit.release_year == ReleaseYear(2010)
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_empty_results_raise_not_found(self, mock_get, client):
def test_empty_results_returns_empty_list(self, mock_get, client):
mock_get.return_value = _ok_response({"results": []})
with pytest.raises(TMDBNotFoundError):
client.search_multi("nothing")
assert client.search_movies("nothing") == []
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_missing_release_date_yields_none_year(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"id": 1, "title": "X"}]}
)
results = client.search_movies("X")
assert results[0].release_year is None
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_malformed_year_yields_none(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"id": 1, "title": "X", "release_date": "soon"}]}
)
results = client.search_movies("X")
assert results[0].release_year is None
# --------------------------------------------------------------------------- #
# search_shows #
# --------------------------------------------------------------------------- #
class TestSearchShows:
@pytest.mark.parametrize("bad", ["", None, 123])
def test_invalid_query_raises_value_error(self, client, bad):
with pytest.raises(ValueError):
client.search_shows(bad)
def test_query_too_long(self, client):
with pytest.raises(ValueError, match="too long"):
client.search_shows("a" * 501)
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_success_wraps_vos(self, mock_get, client):
mock_get.return_value = _ok_response(
{
"results": [
{
"id": 84958,
"name": "Foundation",
"first_air_date": "2021-09-24",
}
]
}
)
results = client.search_shows("Foundation")
assert len(results) == 1
hit = results[0]
assert hit.tmdb_id == TmdbId(84958)
assert hit.name == "Foundation"
assert hit.first_air_year == 2021
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_empty_results_returns_empty_list(self, mock_get, client):
mock_get.return_value = _ok_response({"results": []})
assert client.search_shows("nothing") == []
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_missing_first_air_date_yields_none_year(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"id": 1, "name": "X"}]}
)
results = client.search_shows("X")
assert results[0].first_air_year is None
# --------------------------------------------------------------------------- #
@@ -193,97 +268,6 @@ class TestGetExternalIds:
assert result["imdb_id"] == "tt0903747"
# --------------------------------------------------------------------------- #
# search_media (composite) #
# --------------------------------------------------------------------------- #
class TestSearchMedia:
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_happy_path_movie(self, mock_get, client):
# First call → /search/multi ; second → /movie/X/external_ids
mock_get.side_effect = [
_ok_response(
{
"results": [
{
"id": 27205,
"media_type": "movie",
"title": "Inception",
"overview": "...",
"release_date": "2010-07-15",
"poster_path": "/x.jpg",
"vote_average": 8.4,
}
]
}
),
_ok_response({"imdb_id": "tt1375666"}),
]
result = client.search_media("Inception")
assert isinstance(result, MediaResult)
assert result.title == "Inception"
assert result.imdb_id == "tt1375666"
assert result.media_type == "movie"
assert result.vote_average == 8.4
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_tv_uses_name_field(self, mock_get, client):
mock_get.side_effect = [
_ok_response(
{"results": [{"id": 1396, "media_type": "tv", "name": "Breaking Bad"}]}
),
_ok_response({"imdb_id": "tt0903747"}),
]
result = client.search_media("Breaking Bad")
assert result.title == "Breaking Bad"
assert result.media_type == "tv"
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_person_result_skipped_uses_next(self, mock_get, client):
# First result is a person → falls through to second result.
mock_get.side_effect = [
_ok_response(
{
"results": [
{"id": 1, "media_type": "person", "name": "X"},
{"id": 2, "media_type": "movie", "title": "Y"},
]
}
),
_ok_response({"imdb_id": "tt7654321"}),
]
result = client.search_media("Y")
assert result.title == "Y"
assert result.media_type == "movie"
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_only_person_result_raises_not_found(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"id": 1, "media_type": "person", "name": "X"}]}
)
with pytest.raises(TMDBNotFoundError):
client.search_media("X")
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_malformed_top_result_raises(self, mock_get, client):
mock_get.return_value = _ok_response(
{"results": [{"title": "no id or media_type"}]}
)
with pytest.raises(TMDBAPIError, match="Invalid"):
client.search_media("X")
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_external_ids_failure_returns_result_without_imdb(self, mock_get, client):
# Second call (external IDs) fails — the search should still succeed.
mock_get.side_effect = [
_ok_response({"results": [{"id": 1, "media_type": "movie", "title": "X"}]}),
Timeout("slow"),
]
result = client.search_media("X")
assert result.imdb_id is None
# --------------------------------------------------------------------------- #
# Details endpoints #
# --------------------------------------------------------------------------- #
@@ -333,10 +317,10 @@ class TestGetTvShowInfo:
info = client.get_tv_show_info(84958)
assert info.tmdb_id == 84958
assert info.imdb_id == "tt0804484"
assert info.tmdb_id == TmdbId(84958)
assert info.imdb_id == ImdbId("tt0804484")
assert info.name == "Foundation"
assert info.status == "Returning Series"
assert info.status == ShowStatus.RETURNING_SERIES
assert len(info.seasons) == 2
assert info.seasons[0].number == 1
assert info.seasons[0].episode_count == 10
@@ -353,11 +337,12 @@ class TestGetTvShowInfo:
"seasons": [],
}
),
_ok_response({}), # external_ids without imdb_id
_ok_response({}),
]
info = client.get_tv_show_info(1)
assert info.imdb_id is None
assert info.seasons == ()
assert info.status == ShowStatus.ENDED
class TestGetMovieInfo:
@@ -378,10 +363,10 @@ class TestGetMovieInfo:
info = client.get_movie_info(27205)
assert info.tmdb_id == 27205
assert info.imdb_id == "tt1375666"
assert info.title == "Inception"
assert info.release_year == 2010
assert info.tmdb_id == TmdbId(27205)
assert info.imdb_id == ImdbId("tt1375666")
assert info.title == MovieTitle("Inception")
assert info.release_year == ReleaseYear(2010)
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
def test_missing_imdb_id_becomes_none(self, mock_get, client):
@@ -397,7 +382,7 @@ class TestGetMovieInfo:
]
info = client.get_movie_info(1)
assert info.imdb_id is None
assert info.release_year == 2024
assert info.release_year == ReleaseYear(2024)
class TestIsConfigured:
tests/infrastructure/api/test_tmdb_dto.py
+11 -8
View File
@@ -12,6 +12,9 @@ from datetime import date
import pytest
from alfred.domain.movies.value_objects import MovieTitle, ReleaseYear
from alfred.domain.shared.value_objects import ImdbId, TmdbId
from alfred.domain.tv_shows.value_objects import ShowStatus
from alfred.infrastructure.api.tmdb.dto import (
TmdbMovieInfo,
TmdbSeasonInfo,
@@ -42,10 +45,10 @@ class TestParseTvShowInfoHappyPath:
today=REF_DATE,
)
assert info == TmdbShowInfo(
tmdb_id=84958,
imdb_id="tt0804484",
tmdb_id=TmdbId(84958),
imdb_id=ImdbId("tt0804484"),
name="Foundation",
status="Returning Series",
status=ShowStatus.RETURNING_SERIES,
seasons=(),
)
@@ -192,10 +195,10 @@ class TestParseMovieInfoHappyPath:
{"imdb_id": "tt1375666"},
)
assert info == TmdbMovieInfo(
tmdb_id=27205,
imdb_id="tt1375666",
title="Inception",
release_year=2010,
tmdb_id=TmdbId(27205),
imdb_id=ImdbId("tt1375666"),
title=MovieTitle("Inception"),
release_year=ReleaseYear(2010),
)
def test_release_year_extracted_from_release_date(self):
@@ -203,7 +206,7 @@ class TestParseMovieInfoHappyPath:
_movie_details(release_date="1999-03-31"),
{},
)
assert info.release_year == 1999
assert info.release_year == ReleaseYear(1999)
class TestParseMovieInfoImdb:
tests/infrastructure/persistence/dot_alfred/v2/conftest.py
+4 -4
View File
@@ -24,7 +24,7 @@ from alfred.domain.releases.entities import (
from alfred.domain.releases.value_objects import EpisodeRange, ReleaseMode
from alfred.domain.shared.media import AudioTrack, SubtitleTrack
from alfred.domain.shared.value_objects import FilePath, ImdbId, TmdbId
from alfred.domain.tv_shows.value_objects import EpisodeNumber, SeasonNumber
from alfred.domain.tv_shows.value_objects import EpisodeNumber, SeasonNumber, ShowStatus
from alfred.infrastructure.api.tmdb.dto import TmdbSeasonInfo, TmdbShowInfo
@@ -164,10 +164,10 @@ def inception_release() -> MovieRelease:
def foundation_tmdb_info() -> TmdbShowInfo:
"""Foundation TMDB cache snapshot — 3 seasons, S03 not yet aired."""
return TmdbShowInfo(
tmdb_id=84958,
imdb_id="tt0804484",
tmdb_id=TmdbId(84958),
imdb_id=ImdbId("tt0804484"),
name="Foundation",
status="Returning Series",
status=ShowStatus.RETURNING_SERIES,
seasons=(
TmdbSeasonInfo(number=1, episode_count=10, aired=True),
TmdbSeasonInfo(number=2, episode_count=10, aired=True),