connections

Connection classes for managing provider service connections.

This module defines the base connection classes that handle initialization and authentication with music provider services. Connections manage the underlying API clients and authentication state.

CLASS	DESCRIPTION
Connection	Abstract base class for provider connections.
AuthenticatedConnection	Connection class for providers that require user authentication.

Connection

Connection(
    options: dict | None = None,
    logger: "LoggerLike" = ConsoleLogger(),
)

Abstract base class for provider connections.

A connection manages the lifecycle of connecting to a music provider's service, including initialization and URL validation. Subclasses must implement the abstract methods to handle provider-specific logic.

ATTRIBUTE	DESCRIPTION
_default_options	Default configuration options for the connection. TYPE: dict
_initialized	Whether the connection has been initialized. TYPE: bool
_client	The underlying API client instance. TYPE: object

Initialize the connection with options and a logger.

PARAMETER	DESCRIPTION
options	Configuration options to merge with defaults. TYPE: dict | None DEFAULT: None
logger	Logger instance for logging messages. TYPE: 'LoggerLike' DEFAULT: ConsoleLogger()

Source code in src/downmixer/providers/connections.py

def __init__(
    self, options: Optional[dict] = None, logger: "LoggerLike" = ConsoleLogger()
):
    """Initialize the connection with options and a logger.

    Args:
        options: Configuration options to merge with defaults.
        logger: Logger instance for logging messages.
    """
    self.options = utils.merge_dicts_with_priority(self._default_options, options)
    self.logger = logger

options instance-attribute

options = merge_dicts_with_priority(
    _default_options, options
)

client property

client

The underlying API client instance.

ready property

ready: bool

Whether the connection is ready to be used.

initialize abstractmethod

initialize() -> bool

Initializes the connection to the provider. Should set the initialized attribute to True if successful.

Source code in src/downmixer/providers/connections.py

@abc.abstractmethod
def initialize(self) -> bool:
    """Initializes the connection to the provider. Should set the `initialized` attribute to True if successful."""
    raise NotImplementedError

check_valid_url abstractmethod

check_valid_url(
    url: str, type_filter: list[ResourceType] = None
) -> bool

Check if a URL or URI is valid for this provider.

PARAMETER	DESCRIPTION
url	The URL or URI to validate. TYPE: str
type_filter	Optional list of resource types to filter against. If provided, the URL must match one of these types to be valid. TYPE: list[ResourceType] DEFAULT: None

RETURNS	DESCRIPTION
bool	True if the URL is valid for this provider, False otherwise.

Source code in src/downmixer/providers/connections.py

@abc.abstractmethod
def check_valid_url(self, url: str, type_filter: list[ResourceType] = None) -> bool:
    """Check if a URL or URI is valid for this provider.

    Args:
        url: The URL or URI to validate.
        type_filter: Optional list of resource types to filter against.
            If provided, the URL must match one of these types to be valid.

    Returns:
        True if the URL is valid for this provider, False otherwise.
    """
    raise NotImplementedError

get_resource_type abstractmethod

get_resource_type(value: str) -> ResourceType | None

Determine the resource type of a URL or URI for this provider.

PARAMETER	DESCRIPTION
value	The URL or URI to analyze. TYPE: str

RETURNS	DESCRIPTION
ResourceType | None	The ResourceType if recognized, or None if the URL is not valid.

Source code in src/downmixer/providers/connections.py

@abc.abstractmethod
def get_resource_type(self, value: str) -> ResourceType | None:
    """Determine the resource type of a URL or URI for this provider.

    Args:
        value: The URL or URI to analyze.

    Returns:
        The ResourceType if recognized, or None if the URL is not valid.
    """
    raise NotImplementedError

AuthenticatedConnection

AuthenticatedConnection(
    options: dict | None = None,
    logger: "LoggerLike" = ConsoleLogger(),
)

Bases: Connection

Connection class for providers that require user authentication.

Extends Connection to add authentication state tracking. The connection is only considered ready when both initialized and authenticated.

ATTRIBUTE	DESCRIPTION
_authenticated	Whether the user has been authenticated. TYPE: bool

Source code in src/downmixer/providers/connections.py

def __init__(
    self, options: Optional[dict] = None, logger: "LoggerLike" = ConsoleLogger()
):
    """Initialize the connection with options and a logger.

    Args:
        options: Configuration options to merge with defaults.
        logger: Logger instance for logging messages.
    """
    self.options = utils.merge_dicts_with_priority(self._default_options, options)
    self.logger = logger

ready property

ready: bool

Whether the connection is initialized and authenticated.

authenticate abstractmethod

authenticate(**kwargs) -> bool

Authenticate the user with the provider.

Should set the _authenticated attribute to True if successful.

RETURNS	DESCRIPTION
bool	True if authentication was successful, False otherwise.

Source code in src/downmixer/providers/connections.py

@abc.abstractmethod
def authenticate(self, **kwargs) -> bool:
    """Authenticate the user with the provider.

    Should set the `_authenticated` attribute to True if successful.

    Returns:
        True if authentication was successful, False otherwise.
    """
    raise NotImplementedError

require_authentication

require_authentication(func)

Decorator to ensure a method is only called when authenticated.

PARAMETER	DESCRIPTION
func	The function to wrap.

RETURNS	DESCRIPTION
	A wrapped function that raises NotAuthenticatedException if
	the connection is not authenticated.

Source code in src/downmixer/providers/connections.py

def require_authentication(self, func):
    """Decorator to ensure a method is only called when authenticated.

    Args:
        func: The function to wrap.

    Returns:
        A wrapped function that raises NotAuthenticatedException if
        the connection is not authenticated.
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        if not self._authenticated or not self._initialized:
            raise NotAuthenticatedException(
                "This platform requires authentication to perform this action."
            )
        return func(*args, **kwargs)

    return wrapper