discogs

Discogs API client for music metadata retrieval.

This module provides the Discogs-specific implementation for fetching and scoring music releases from the Discogs database.

DiscogsFormat

Bases: TypedDict

Type definition for Discogs format information.

DiscogsRelease

Bases: TypedDict

Type definition for Discogs release from search results.

DiscogsSearchResponse

Bases: TypedDict

Type definition for Discogs search API response.

DiscogsMasterRelease

Bases: TypedDict

Type definition for Discogs master release response.

DiscogsClient

DiscogsClient(
    token,
    console_logger,
    error_logger,
    analytics,
    make_api_request_func,
    *,
    score_release_func,
    cache_service,
    scoring_config,
    config,
    cache_ttl_days=30
)

Bases: BaseApiClient

Discogs API client for fetching music metadata.

Initialize Discogs client.

Parameters:

Name	Type	Description	Default
token	str	Discogs API token	required
console_logger	Logger	Logger for console output	required
error_logger	Logger	Logger for error messages	required
analytics	Analytics	Analytics service for performance tracking	required
make_api_request_func	Callable[..., Awaitable[dict[str, Any] | None]]	Function to make API requests with rate-limiting	required
score_release_func	Callable[..., float]	Function to score releases for originality	required
cache_service	CacheServiceProtocol	Cache service for storing results	required
scoring_config	YearRetrievalConfig	Year retrieval configuration with scoring rules	required
config	AppConfig	Typed application configuration	required
cache_ttl_days	int	Cache TTL in days	30

Source code in src/services/api/discogs.py

def __init__(
    self,
    token: str,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    analytics: Analytics,
    make_api_request_func: Callable[..., Awaitable[dict[str, Any] | None]],
    *,
    score_release_func: Callable[..., float],
    cache_service: CacheServiceProtocol,
    scoring_config: YearRetrievalConfig,
    config: AppConfig,
    cache_ttl_days: int = 30,
) -> None:
    """Initialize Discogs client.

    Args:
        token: Discogs API token
        console_logger: Logger for console output
        error_logger: Logger for error messages
        analytics: Analytics service for performance tracking
        make_api_request_func: Function to make API requests with rate-limiting
        score_release_func: Function to score releases for originality
        cache_service: Cache service for storing results
        scoring_config: Year retrieval configuration with scoring rules
        config: Typed application configuration
        cache_ttl_days: Cache TTL in days

    """
    super().__init__(console_logger, error_logger)
    self.analytics = analytics
    self.token = token
    self._make_api_request = make_api_request_func
    self._score_original_release = score_release_func
    self.cache_service = cache_service
    self.scoring_config = scoring_config
    self.config = config
    self.cache_ttl_days = cache_ttl_days

get_year_from_discogs async

get_year_from_discogs(artist, album)

Get year from Discogs (for backward compatibility).

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
str | None	Year string or None

Source code in src/services/api/discogs.py

@track_instance_method("discogs_year_search")
async def get_year_from_discogs(self, artist: str, album: str) -> str | None:
    """Get year from Discogs (for backward compatibility).

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Year string or None

    """
    releases = await self.get_scored_releases(
        self._normalize_name(artist),
        self._normalize_name(album),
        None,
        artist_orig=artist,
        album_orig=album,
    )

    if releases:
        # Return the year from the highest scored release
        best_release = max(releases, key=lambda x: x["score"])
        year = best_release.get("year")
        return str(year) if year is not None else None

    return None

get_scored_releases async

get_scored_releases(
    artist_norm,
    album_norm,
    artist_region,
    *,
    artist_orig=None,
    album_orig=None
)

Retrieve and score releases from Discogs.

Parameters:

Name	Type	Description	Default
artist_norm	str	Normalized artist name	required
album_norm	str	Normalized album name	required
artist_region	str | None	Artist's region for scoring	required
artist_orig	str | None	Original artist name (before normalization)	None
album_orig	str | None	Original album name (before normalization)	None

Returns:

Type	Description
list[ScoredRelease]	List of scored releases sorted by score

Source code in src/services/api/discogs.py

@track_instance_method("discogs_release_search")
async def get_scored_releases(
    self,
    artist_norm: str,
    album_norm: str,
    artist_region: str | None,
    *,
    artist_orig: str | None = None,
    album_orig: str | None = None,
) -> list[ScoredRelease]:
    """Retrieve and score releases from Discogs.

    Args:
        artist_norm: Normalized artist name
        album_norm: Normalized album name
        artist_region: Artist's region for scoring
        artist_orig: Original artist name (before normalization)
        album_orig: Original album name (before normalization)

    Returns:
        List of scored releases sorted by score

    """
    cache_key = f"discogs_{artist_norm}_{album_norm}"
    cache_ttl_seconds = self.cache_ttl_days * 86400

    # Check cache first
    cached_releases = await self._get_cached_discogs_releases(cache_key)
    if cached_releases is not None:
        return cached_releases

    try:
        # Make search request
        discogs_response = await self._make_discogs_search_request(artist_norm, album_norm, artist_orig, album_orig)

        if discogs_response is None:
            await self.cache_service.set_async(cache_key, [], ttl=cache_ttl_seconds)
            return []

        results = discogs_response.get("results", [])

        # Get reissue keywords
        reissue_keywords = self._get_reissue_keywords()

        # Process results
        scored_releases: list[ScoredRelease] = await self._process_discogs_results(
            results,
            artist_norm,
            album_norm,
            artist_region=artist_region,
            reissue_keywords=reissue_keywords,
        )

        # Cache results
        await self.cache_service.set_async(cache_key, scored_releases, ttl=cache_ttl_seconds)

    except (OSError, ValueError, RuntimeError, KeyError, TypeError, AttributeError) as e:
        self.error_logger.exception("Error fetching from Discogs for '%s - %s': %s", artist_norm, album_norm, e)
        return []

    return sorted(scored_releases, key=lambda x: x["score"], reverse=True)