Coverage for wallhavenapi/wallhavenapi.py: 89%
226 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-14 20:30 +0000
1"""
2Wallhaven API v1 Python Wrapper
4This module provides a client interface to the Wallhaven.cc API v1.
5It enables users to search, retrieve, and download wallpapers,
6as well as manage collections.
8Author: Ray Cadle
9License: MIT
10"""
12import requests
13import os
14import random
15import string
16import time
17from enum import Enum
18from typing import Tuple, Dict, List, Optional, Union, Any
20# ---------- Enums ----------
22class Purity(Enum):
23 """Defines safety filters for wallpaper content used to filter SFW, sketchy, or NSFW results."""
24 sfw = "sfw"
25 sketchy = "sketchy"
26 nsfw = "nsfw"
29class Category(Enum):
30 """Defines wallpaper categories for filtering search results."""
31 general = "general"
32 anime = "anime"
33 people = "people"
36class Sorting(Enum):
37 """Defines sorting options for search results returned by the API."""
38 date_added = "date_added"
39 relevance = "relevance"
40 random = "random"
41 views = "views"
42 favorites = "favorites"
43 toplist = "toplist"
46class Order(Enum):
47 """Defines the ordering direction for sorting (ascending or descending). Default: desc"""
48 desc = "desc"
49 asc = "asc"
52class TopRange(Enum):
53 """Defines time-based ranges used when sorting by 'toplist'."""
54 one_day = "1d"
55 three_days = "3d"
56 one_week = "1w"
57 one_month = "1M"
58 three_months = "3M"
59 six_months = "6M"
60 one_year = "1y"
63class Color(Enum):
64 """Provides a list of predefined color hex codes for filtering wallpapers by dominant color."""
65 lonestar = "660000"
66 red_berry = "990000"
67 guardsman_red = "cc0000"
68 persian_red = "cc3333"
69 french_rose = "ea4c88"
70 plum = "993399"
71 royal_purple = "663399"
72 sapphire = "333399"
73 science_blue = "0066cc"
74 pacific_blue = "0099cc"
75 downy = "66cccc"
76 atlantis = "77cc33"
77 limeade = "669900"
78 verdun_green = "336600"
79 verdun_green_2 = "666600"
80 olive = "999900"
81 earls_green = "cccc33"
82 yellow = "ffff00"
83 sunglow = "ffcc33"
84 orange_peel = "ff9900"
85 blaze_orange = "ff6600"
86 tuscany = "cc6633"
87 potters_clay = "996633"
88 nutmeg_wood_finish = "663300"
89 black = "000000"
90 dusty_gray = "999999"
91 silver = "cccccc"
92 white = "ffffff"
93 gun_powder = "424153"
96class Type(Enum):
97 """Defines supported image formats for filtering search results."""
98 jpeg = "jpeg"
99 jpg = "jpg"
100 png = "png"
103# ---------- Utilities ----------
105class Seed:
106 """Utility class for generating random alphanumeric seeds."""
107 @staticmethod
108 def generate() -> str:
109 """
110 Generate a random 6-character alphanumeric seed string.
112 Returns:
113 str: Random seed composed of letters and digits.
114 """
115 return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(6))
118# ---------- Exceptions ----------
120class RequestsLimitError(Exception):
121 """
122 Raised when the API request limit (HTTP 429) is exceeded.
124 Attributes:
125 message (str, optional): A custom error message that overrides the default.
126 status_code (int): HTTP status code received from the API.
127 """
128 def __init__(
129 self,
130 message: Optional[str] = None,
131 status_code: int = 429
132 ):
133 self.status_code = status_code
134 default_msg: str = "You have exceeded the requests limit. Please try later."
135 super().__init__(message or default_msg)
138class ApiKeyError(Exception):
139 """
140 Raised when an invalid, missing, or unauthorized API key is used.
142 Attributes:
143 message (str, optional): A custom error message that overrides the default.
144 status_code (int): HTTP status code received from the API.
145 """
146 def __init__(
147 self,
148 message: Optional[str] = None,
149 status_code: int = 401
150 ):
151 self.status_code = status_code
152 default_msg: str = "Bad API key. Check it please."
153 super().__init__(message or default_msg)
156class NoWallpaperError(Exception):
157 """
158 Raised when no wallpaper with the specified ID exists.
160 Attributes:
161 wallpaper_id (str): ID of the wallpaper that was not found.
162 message (str, optional): A custom error message that overrides the default.
163 status_code (int): HTTP status code received from the API.
164 """
165 def __init__(
166 self,
167 wallpaper_id: str,
168 message: Optional[str] = None,
169 status_code: int = 404
170 ):
171 self.wallpaper_id = wallpaper_id
172 self.status_code = status_code
173 default_msg: str = f"No wallpaper with id {wallpaper_id}"
174 super().__init__(message or default_msg)
177class UnhandledException(Exception):
178 """
179 Raised for any unhandled API errors or unexpected HTTP responses.
181 Attributes:
182 message (str, optional): A custom error message that overrides the default.
183 status_code (int, optional): HTTP status code if available.
184 """
185 def __init__(
186 self,
187 message: Optional[str] = None,
188 status_code: Optional[int] = None
189 ):
190 self.status_code = status_code
191 default_msg: str = "Something went wrong. Please submit this issue to https://github.com/raycadle/WallhavenAPI/issues."
192 super().__init__(message or default_msg)
195# ---------- API Client Class ----------
197class WallhavenAPI:
198 """
199 Main interface class for interacting with the Wallhaven.cc API v1.
201 This class handles requests, retries on failures, search queries,
202 downloads, and account-specific endpoints like collections or settings.
204 Attributes:
205 api_key (str, optional): Wallhaven API key (optional for some endpoints).
206 verify_connection (bool): Whether to verify SSL certificates.
207 base_url (str): The base API endpoint URL.
208 timeout (tuple of integers): Request timeout settings.
209 requestslimit_timeout (tuple of integers, optional): Retry configuration on rate limits.
210 proxies (dictionary of strings): HTTP/HTTPS proxy settings.
211 """
212 def __init__(
213 self,
214 api_key: Optional[str] = None,
215 verify_connection: bool = True,
216 base_url: str = "https://wallhaven.cc/api/v1",
217 timeout: Tuple[int, int] = (2, 5),
218 requestslimit_timeout: Optional[Tuple[int, int]] = None,
219 proxies: Dict[str, str] = None
220 ):
221 self.api_key = api_key
222 self.verify_connection = verify_connection
223 self.base_url = base_url
224 self.timeout = timeout
225 self.requestslimit_timeout = requestslimit_timeout
226 self.proxies = proxies or {}
228 def _request(
229 self,
230 to_json: bool,
231 **kwargs: Any
232 ) -> Union[dict, requests.Response]:
233 """
234 Internal method to perform HTTP requests with retry and error handling.
236 Args:
237 to_json (bool): Whether to return the response as JSON.
238 **kwargs: Parameters passed to requests.request.
240 Returns:
241 dict or requests.Response: Parsed JSON response or raw response.
243 Raises:
244 RequestsLimitError: If rate-limited and retries are exhausted.
245 ApiKeyError: If API key is invalid.
246 UnhandledException: For all other unexpected issues.
247 """
248 max_retries = self.requestslimit_timeout[0] if self.requestslimit_timeout else 1
249 delay = self.requestslimit_timeout[1] if self.requestslimit_timeout else 0
251 for attempt in range(max_retries):
252 # Add API key to query params if available
253 if self.api_key:
254 kwargs.setdefault("params", {})["apikey"] = self.api_key
256 # Apply request configuration
257 kwargs.setdefault("timeout", self.timeout)
258 kwargs.setdefault("verify", self.verify_connection)
259 kwargs.setdefault("proxies", self.proxies)
261 # Send the request
262 try:
263 response = requests.request(**kwargs)
264 except requests.RequestException as e:
265 if attempt == max_retries - 1:
266 raise UnhandledException(message=f"Request failed: {str(e)}")
267 time.sleep(delay)
268 continue
270 status_code = response.status_code
272 # Handle rate limiting (retry if needed)
273 if status_code == 429:
274 if attempt == max_retries - 1:
275 raise RequestsLimitError(status_code=status_code)
276 time.sleep(delay)
277 continue
279 # Handle invalid API key
280 if status_code == 401:
281 raise ApiKeyError(status_code=status_code)
283 # Handle 404 (let caller interpret if needed)
284 if status_code == 404:
285 raise UnhandledException(
286 message=f"404 Not Found for URL: {response.url}",
287 status_code=status_code
288 )
290 # Handle any other non-200 status codes
291 if status_code != 200:
292 raise UnhandledException(
293 message=f"Unexpected status code {status_code} for URL: {response.url}",
294 status_code=status_code
295 )
297 # Return JSON or raw response
298 if to_json:
299 try:
300 return response.json()
301 except Exception as e:
302 raise UnhandledException(
303 message=f"JSON decode error: {str(e)}",
304 status_code=status_code
305 )
307 return response
309 # If somehow loop ends without return or raise, raise generic error
310 raise UnhandledException(message="Request failed after all retry attempts.")
312 def _raw_request(
313 self,
314 url: str
315 ) -> requests.Response:
316 """
317 Perform a raw GET request with retry and error handling, respecting client settings.
319 Args:
320 url (str): The full URL of the resource to download.
322 Returns:
323 requests.Response: The HTTP response with streamed content.
325 Raises:
326 RequestsLimitError: If too many requests and retries exhausted.
327 UnhandledException: For unexpected HTTP errors.
328 """
329 max_retries = self.requestslimit_timeout[0] if self.requestslimit_timeout else 1
330 delay = self.requestslimit_timeout[1] if self.requestslimit_timeout else 0
332 for attempt in range(max_retries):
334 # Send the request
335 try:
336 response = requests.get(
337 url,
338 stream=True,
339 timeout=self.timeout,
340 verify=self.verify_connection,
341 proxies=self.proxies,
342 )
343 if response.status_code == 200:
344 return response
345 elif response.status_code == 429:
346 if attempt == max_retries - 1:
347 raise RequestsLimitError()
348 time.sleep(delay)
349 continue
350 else:
351 raise UnhandledException(
352 message=f"Unexpected status code {response.status_code} for URL: {url}",
353 status_code=response.status_code,
354 )
355 except requests.RequestException as e:
356 # Network-related errors or connection issues
357 if attempt == max_retries - 1:
358 raise UnhandledException(message=f"Request failed: {str(e)}")
359 time.sleep(delay)
361 # If somehow loop ends without return or raise, raise generic error
362 raise UnhandledException(message="Failed to download after multiple attempts.")
364 def _format_url(
365 self,
366 *args: Union[str, int]
367 ) -> str:
368 """
369 Build a formatted API endpoint URL by appending path components.
371 Args:
372 *args (str or int): Path components to join to the base URL.
374 Returns:
375 str: Full URL to the API endpoint.
376 """
377 url = self.base_url.rstrip("/") + "/"
378 return url + "/".join(map(str, args))
380 @staticmethod
381 def _category(
382 general: bool = True,
383 anime: bool = True,
384 people: bool = False
385 ) -> str:
386 """
387 Convert category flags to API-compatible string format.
389 Args:
390 general (bool): Include general wallpapers.
391 anime (bool): Include anime wallpapers.
392 people (bool): Include people wallpapers.
394 Returns:
395 str: Category format string, e.g., '110'.
396 """
397 return f"{int(general)}{int(anime)}{int(people)}"
399 @staticmethod
400 def _purity(
401 sfw: bool = True,
402 sketchy: bool = True,
403 nsfw: bool = False
404 ) -> str:
405 """
406 Convert purity flags to API-compatible string format.
408 Args:
409 sfw (bool): Include safe-for-work content.
410 sketchy (bool): Include sketchy content.
411 nsfw (bool): Include not-safe-for-work content.
413 Returns:
414 str: Purity format string, e.g., '110'.
415 """
416 return f"{int(sfw)}{int(sketchy)}{int(nsfw)}"
418 @staticmethod
419 def _format_dimensions(dims: Union[Tuple[int, int], List[Tuple[int, int]]]) -> str:
420 """
421 Format a single dimension tuple or a list of tuples into
422 a comma-separated string suitable for the API.
424 Args:
425 dims (tuple of integers or list of tuples of integers):
426 A single (width, height) tuple or list of such tuples.
428 Returns:
429 str: A string formatted as "WxH,WxH,..."
430 """
431 dims = dims if isinstance(dims, list) else [dims]
432 return ",".join(f"{w}x{h}" for w, h in dims)
434 def search(
435 self,
436 q: Optional[str] = None,
437 categories: Optional[Union[Category, List[Category]]] = None,
438 purities: Optional[Union[Purity, List[Purity]]] = None,
439 sorting: Optional[Sorting] = None,
440 order: Optional[Order] = None,
441 top_range: Optional[TopRange] = None,
442 atleast: Optional[Tuple[int, int]] = None,
443 resolutions: Optional[Union[Tuple[int, int], List[Tuple[int, int]]]] = None,
444 ratios: Optional[Union[Tuple[int, int], List[Tuple[int, int]]]] = None,
445 colors: Optional[Color] = None,
446 page: Optional[int] = None,
447 seed: Optional[str] = None
448 ) -> dict:
449 """
450 Search for wallpapers using various filters and parameters.
452 Args:
453 q (str, optional): Query string (e.g., keywords or tags).
454 categories (list or Category, optional): Categories to include.
455 purities (list or Purity, optional): Purity filters (SFW, NSFW, etc.).
456 sorting (Sorting, optional): How to sort the results.
457 order (Order, optional): Sort direction (asc or desc).
458 top_range (TopRange, optional): Time range for toplist sorting.
459 atleast (tuple of integers, optional): Minimum resolution (width, height).
460 resolutions (tuple of integers or list of tuples of integers, optional): Exact resolutions.
461 ratios (tuple of integers or list of tuples of integers, optional): Screen ratios (e.g., 16:9).
462 colors (Color, optional): Dominant color to filter by.
463 page (int, optional): Page number of results.
464 seed (str, optional): Seed for reproducible random results.
466 Returns:
467 dict: JSON response from Wallhaven API.
468 """
469 params: Dict[str, str] = {}
470 if q: params["q"] = q
471 if categories:
472 categories = categories if isinstance(categories, list) else [categories]
473 params["categories"] = self._category(
474 Category.general in categories,
475 Category.anime in categories,
476 Category.people in categories,
477 )
478 if purities:
479 purities = purities if isinstance(purities, list) else [purities]
480 params["purity"] = self._purity(
481 Purity.sfw in purities,
482 Purity.sketchy in purities,
483 Purity.nsfw in purities,
484 )
485 if sorting: params["sorting"] = sorting.value
486 if order: params["order"] = order.value
487 if top_range: params["topRange"] = top_range.value
488 if atleast: params["atleast"] = f"{atleast[0]}x{atleast[1]}"
489 if resolutions: params["resolutions"] = self._format_dimensions(resolutions)
490 if ratios: params["ratios"] = self._format_dimensions(ratios)
491 if colors: params["colors"] = colors.value
492 if page: params["page"] = str(page)
493 if seed: params["seed"] = seed
495 return self._request(True, method="get", url=self._format_url("search"), params=params)
497 def wallpaper(
498 self,
499 wallpaper_id: str
500 ) -> dict:
501 """
502 Retrieve metadata for a specific wallpaper by ID.
504 Args:
505 wallpaper_id (str): The unique ID of the wallpaper.
507 Returns:
508 dict: Metadata about the wallpaper.
510 Raises:
511 NoWallpaperError: If the wallpaper is not found.
512 """
513 try:
514 return self._request(True, method="get", url=self._format_url("w", wallpaper_id))
515 except UnhandledException as e:
516 # If the error was due to a 404, convert it to a NoWallpaperError
517 if e.status_code == 404:
518 raise NoWallpaperError(wallpaper_id)
519 raise # Re-raise other unhandled exceptions
521 def is_wallpaper_exists(
522 self,
523 wallpaper_id: str
524 ) -> bool:
525 """
526 Check if a wallpaper exists on Wallhaven.
528 Args:
529 wallpaper_id (str): The wallpaper ID to check.
531 Returns:
532 bool: True if wallpaper exists, False otherwise.
533 """
534 try:
535 self.wallpaper(wallpaper_id)
536 return True
537 except NoWallpaperError:
538 return False
540 def download_wallpaper(
541 self,
542 wallpaper_id: str,
543 file_path: Optional[str],
544 chunk_size: int = 4096
545 ) -> Union[str, bytes]:
546 """
547 Download wallpaper by ID.
549 Args:
550 wallpaper_id (str): Wallpaper ID.
551 file_path (str, optional): Path where image should be saved. If None, returns binary content.
552 chunk_size (int): Stream chunk size.
554 Returns:
555 str or bytes: Saved path or raw content.
556 """
557 wallpaper_data = self.wallpaper(wallpaper_id)
558 wallpaper = self._raw_request(wallpaper_data["data"]["path"])
560 if file_path:
561 save_path = os.path.abspath(file_path)
562 os.makedirs(os.path.dirname(save_path), exist_ok=True)
563 with open(save_path, "wb") as f:
564 for chunk in wallpaper.iter_content(chunk_size=chunk_size):
565 if chunk:
566 f.write(chunk)
567 return save_path
569 return b"".join(wallpaper.iter_content(chunk_size=chunk_size))
571 def tag(
572 self,
573 tag_id: Union[str, int]
574 ) -> dict:
575 """
576 Retrieve tag details by tag ID.
578 Args:
579 tag_id (str or int): ID of the tag to retrieve.
581 Returns:
582 dict: Tag metadata.
583 """
584 return self._request(
585 True,
586 method="get",
587 url=self._format_url("tag", tag_id)
588 )
590 def settings(self) -> dict:
591 """
592 Retrieve account settings (requires valid API key).
594 Returns:
595 dict: User settings as provided by Wallhaven.
597 Raises:
598 ApiKeyError: If API key is missing or invalid.
599 """
600 if self.api_key is None:
601 raise ApiKeyError("API key required to retrieve settings.")
602 return self._request(
603 True,
604 method="get",
605 url=self._format_url("settings")
606 )
608 def my_collections(self) -> dict:
609 """
610 Get personal collections for the current user (requires API key).
612 Returns:
613 dict: User's collections.
615 Raises:
616 ApiKeyError: If API key is missing.
617 """
618 if self.api_key is None:
619 raise ApiKeyError("API key required to retrieve collections.")
620 return self._request(
621 True,
622 method="get",
623 url=self._format_url("collections")
624 )
626 def user_collections(
627 self,
628 user_name: str
629 ) -> dict:
630 """
631 Retrieve public collections of another Wallhaven user.
633 Args:
634 user_name (str): The username whose collections to fetch.
636 Returns:
637 dict: Public collections for that user.
638 """
639 return self._request(
640 True,
641 method="get",
642 url=self._format_url("collections", user_name)
643 )
645 def collection_wallpapers(
646 self,
647 user_name: str,
648 collection_id: Union[str, int],
649 page: Optional[int] = None
650 ) -> dict:
651 """
652 Fetch wallpapers from a specific user's collection.
654 Args:
655 user_name (str): Username who owns the collection.
656 collection_id (str or int): The collection's ID.
657 page (int, optional): Page number of results.
659 Returns:
660 dict: Wallpapers from the collection.
661 """
662 params = {"page": str(page)} if page is not None else {}
663 return self._request(
664 True,
665 method="get",
666 url=self._format_url("collections", user_name, collection_id),
667 params=params
668 )