"""
ThePornDB API client with retry logic and error handling.
Adapted from the movie-nfo project with enhancements for MCP service requirements.
"""
import logging
import requests
from requests.exceptions import RequestException
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
before_sleep_log
)
from typing import Optional
from .models import SearchResponse, DetailResponse, MediaData
from .config import Config, API_BASE_URL
logger = logging.getLogger(__name__)
class RateLimitError(Exception):
"""Raised when ThePornDB API returns HTTP 429 (rate limit exceeded)."""
pass
class APIError(Exception):
"""Raised when ThePornDB API request fails."""
pass
class ThePornDBAPI:
"""
ThePornDB API client with retry logic and structured error handling.
Features:
- Automatic retry with exponential backoff for rate limit errors
- URL encoding for search terms
- Structured error messages
- Secure logging (never logs token values)
"""
def __init__(self, config: Config):
"""
Initialize API client.
Args:
config: Application configuration
"""
self.config = config
self.session = requests.Session()
logger.info("ThePornDB API client initialized")
def _get_headers(self) -> dict:
"""
Get request headers with authentication.
Returns:
Headers dictionary with Authorization and Accept headers
"""
return {
'Authorization': f'Bearer {self.config.token}',
'Accept': 'application/json'
}
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(RateLimitError),
before_sleep=before_sleep_log(logger, logging.INFO)
)
def _request(self, url: str) -> dict:
"""
Make HTTP request with retry logic for rate limiting.
Args:
url: Full request URL
Returns:
Response JSON as dictionary
Raises:
RateLimitError: HTTP 429 response (will be retried)
APIError: Other HTTP errors or request failures
"""
self.config.log_request(url)
headers = self._get_headers()
self.config.log_request_debug(url, headers)
try:
response = self.session.get(url, headers=headers, timeout=15)
# Check for rate limit error
if response.status_code == 429:
logger.warning("Rate limit hit (429), retrying with exponential backoff...")
raise RateLimitError(f"Rate limit exceeded: {url}")
# Check for other HTTP errors
response.raise_for_status()
# Successful response
result = response.json()
if 'data' in result:
result_count = len(result.get('data', []))
self.config.log_response(response.status_code, result_count)
else:
self.config.log_response(response.status_code)
return result
except RequestException as e:
logger.error(f"Request failed: {e}")
raise APIError(f"Failed to fetch data from ThePornDB API: {e}")
def search_data(
self,
search_term: str,
content_type: str,
year: Optional[int] = None
) -> Optional[SearchResponse]:
"""
Search for content by term and optional year.
Args:
search_term: Search query term
content_type: Content type (scenes, movies, jav)
year: Optional year filter
Returns:
SearchResponse with paginated results
Raises:
APIError: If request fails after retries
"""
# URL-encode search term
parsed_term = requests.utils.quote(search_term)
url = f"{API_BASE_URL}/{content_type}?parse={parsed_term}"
if year:
url += f"&year={year}"
try:
response = self._request(url)
return response
except APIError as e:
logger.error(f"Search failed for '{search_term}' in {content_type}: {e}")
return None
def fetch_data(
self,
item_id: str,
content_type: str
) -> Optional[MediaData]:
"""
Fetch detailed content information by ID.
Args:
item_id: Unique content identifier
content_type: Content type (scene, movie, jav)
Returns:
MediaData with complete content details
Raises:
APIError: If request fails after retries
"""
url = f"{API_BASE_URL}/{content_type}/{item_id}"
try:
response = self._request(url)
# API returns structure: { "data": { ... } }
detail_response: DetailResponse = response
return detail_response.get('data')
except APIError as e:
logger.error(f"Fetch failed for {content_type}/{item_id}: {e}")
return None
def search_performers(self, name: str) -> Optional[SearchResponse]:
"""
Search for performers by name.
Args:
name: Performer name to search
Returns:
SearchResponse with performer results
Raises:
APIError: If request fails after retries
"""
parsed_name = requests.utils.quote(name)
url = f"{API_BASE_URL}/performers?parse={parsed_name}"
try:
response = self._request(url)
return response
except APIError as e:
logger.error(f"Performer search failed for '{name}': {e}")
return None
def fetch_performer_details(self, performer_id: str) -> Optional[dict]:
"""
Fetch detailed performer information.
Args:
performer_id: Unique performer identifier
Returns:
Dictionary with complete performer details
Raises:
APIError: If request fails after retries
"""
url = f"{API_BASE_URL}/performers/{performer_id}"
try:
response = self._request(url)
return response.get('data') if 'data' in response else response
except APIError as e:
logger.error(f"Performer details fetch failed for {performer_id}: {e}")
return None
def validate_token(self) -> bool:
"""
Validate API token by making a lightweight API call.
Makes a simple search request to verify token is valid.
Returns:
True if token is valid, False otherwise
"""
try:
# Make a lightweight search request
url = f"{API_BASE_URL}/scenes?parse=test"
response = self._request(url)
logger.info("Token validation successful")
return True
except (APIError, RateLimitError) as e:
logger.error(f"Token validation failed: {e}")
return False
