-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapi_client.py
More file actions
152 lines (115 loc) · 5.17 KB
/
Copy pathapi_client.py
File metadata and controls
152 lines (115 loc) · 5.17 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
"""
api_client.py
Handles all communication with the OMDb API. This module is responsible
for building requests, sending them, and translating raw JSON responses
into Movie objects (or raising/returning clear errors). No CLI or
presentation logic lives here -- see main.py and utils.py for that.
"""
from typing import List, Optional
import requests
import config
from movie import Movie
class ApiClientError(Exception):
"""Raised when the OMDb API cannot be reached or returns a bad response."""
class MovieNotFoundError(Exception):
"""Raised when OMDb successfully responds but finds no matching movie(s)."""
class OmdbApiClient:
"""A small client for interacting with the OMDb API."""
def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None) -> None:
"""
Initialize the API client.
Args:
api_key: The OMDb API key to use. Defaults to the key loaded
in config.py from the environment.
base_url: The OMDb base URL. Defaults to config.OMDB_BASE_URL.
"""
self.api_key = api_key if api_key is not None else config.OMDB_API_KEY
self.base_url = base_url if base_url is not None else config.OMDB_BASE_URL
def _get(self, params: dict) -> dict:
"""
Perform a GET request against the OMDb API with the given params.
Args:
params: Query parameters to send (excluding the API key,
which is added automatically).
Returns:
dict: The parsed JSON response body.
Raises:
ApiClientError: If the API key is missing, the request times
out, a network error occurs, or the response is not
valid JSON / not HTTP 200.
"""
if not self.api_key or not self.api_key.strip():
raise ApiClientError(
"OMDb API key is not configured. Please set the OMDB_API_KEY "
"environment variable (see .env.example)."
)
request_params = dict(params)
request_params["apikey"] = self.api_key
try:
response = requests.get(
self.base_url,
params=request_params,
timeout=config.REQUEST_TIMEOUT,
)
except requests.exceptions.Timeout as exc:
raise ApiClientError("The request to OMDb timed out. Please try again.") from exc
except requests.exceptions.ConnectionError as exc:
raise ApiClientError(
"Could not connect to OMDb. Please check your internet connection."
) from exc
except requests.exceptions.RequestException as exc:
raise ApiClientError(f"An unexpected network error occurred: {exc}") from exc
if response.status_code != 200:
raise ApiClientError(
f"OMDb returned an unexpected HTTP status code: {response.status_code}"
)
try:
data = response.json()
except ValueError as exc:
raise ApiClientError("OMDb returned a response that was not valid JSON.") from exc
if not isinstance(data, dict):
raise ApiClientError("OMDb returned an unexpected response format.")
return data
def search_movies(self, title: str) -> List[Movie]:
"""
Search OMDb for movies matching the given title.
Args:
title: The movie title (or partial title) to search for.
Returns:
list[Movie]: A list of matching Movie objects (basic info only).
Raises:
ValueError: If title is empty or only whitespace.
MovieNotFoundError: If OMDb reports no results were found.
ApiClientError: If the request fails for any other reason.
"""
title = (title or "").strip()
if not title:
raise ValueError("Search title must not be empty.")
data = self._get({"s": title, "type": "movie"})
if data.get("Response") == "False":
error_message = data.get("Error", "No movies found.")
raise MovieNotFoundError(error_message)
results = data.get("Search", [])
if not isinstance(results, list):
raise ApiClientError("OMDb returned malformed search results.")
return [Movie.from_omdb_search_result(item) for item in results]
def get_movie_details(self, imdb_id: str) -> Movie:
"""
Fetch full details for a single movie by its IMDb ID.
Args:
imdb_id: The IMDb identifier (e.g. "tt0816692").
Returns:
Movie: A fully populated Movie object.
Raises:
ValueError: If imdb_id is empty.
MovieNotFoundError: If OMDb cannot find a movie with that ID.
ApiClientError: If the request fails for any other reason.
"""
imdb_id = (imdb_id or "").strip()
if not imdb_id:
raise ValueError("imdb_id must not be empty.")
data = self._get({"i": imdb_id, "plot": "full"})
if data.get("Response") == "False":
error_message = data.get("Error", "Movie details not found.")
raise MovieNotFoundError(error_message)
return Movie.from_omdb_details(data)