tessl/pypi-mopidy
Mopidy is an extensible music server written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Securityby
Pending
The risk profile of this skill
backend-system.mddocs/
Backend System
Mopidy's backend system provides a pluggable architecture for integrating different music sources. Backends act as adapters between Mopidy's core system and external music services or local storage, providing standardized interfaces for library access, playback control, and playlist management.
Capabilities
Backend Base Class
The main backend interface that coordinates different provider types and defines the backend's capabilities.
class Backend:
"""
Base class for Mopidy backends providing music source integration.
Parameters:
- config: Backend configuration
- audio: Audio actor reference
"""
def __init__(self, config, audio): ...
# Backend properties
uri_schemes: list[str] # URI schemes handled by this backend
library: LibraryProvider # Library browsing/searching provider
playback: PlaybackProvider # Playback control provider
playlists: PlaylistsProvider # Playlist management provider
def has_library(self) -> bool:
"""Check if backend provides library functionality."""
...
def has_library_browse(self) -> bool:
"""Check if backend supports library browsing."""
...
def has_playback(self) -> bool:
"""Check if backend provides playback functionality."""
...
def has_playlists(self) -> bool:
"""Check if backend provides playlist functionality."""
...
def on_start(self):
"""Called when backend is started."""
...
def on_stop(self):
"""Called when backend is stopped."""
...Usage example:
class MyBackend(Backend):
def __init__(self, config, audio):
super().__init__(config, audio)
self.uri_schemes = ["myservice"]
self.library = MyLibraryProvider(backend=self)
self.playback = MyPlaybackProvider(audio=audio, backend=self)Library Provider Interface
Provider interface for library browsing, searching, and track lookup functionality.
class LibraryProvider:
"""Base class for library providers handling music browsing and search."""
def __init__(self, backend): ...
def browse(self, uri):
"""
Browse library at given URI.
Parameters:
- uri (str): URI to browse
Returns:
- list[Ref]: References found at URI
"""
...
def search(self, query=None, uris=None, exact=False):
"""
Search library for tracks, artists, and albums.
Parameters:
- query (dict): Search query by field (artist, album, track_name, etc.)
- uris (list[str], optional): URIs to search within
- exact (bool): Whether to perform exact matching
Returns:
- SearchResult: Search results
"""
...
def lookup(self, uris):
"""
Lookup tracks by URI.
Parameters:
- uris (list[str]): Track URIs to lookup
Returns:
- dict[str, list[Track]]: Mapping of URI to track list
"""
...
def refresh(self, uri=None):
"""
Refresh library cache.
Parameters:
- uri (str, optional): Specific URI to refresh
Returns:
- bool: True if successful
"""
...
def get_distinct(self, field, query=None):
"""
Get distinct values for a field.
Parameters:
- field (str): Field name (artist, album, date, etc.)
- query (dict, optional): Query to filter results
Returns:
- set[str]: Set of distinct values
"""
...
def get_images(self, uris):
"""
Get images for given URIs.
Parameters:
- uris (list[str]): URIs to get images for
Returns:
- dict[str, list[Image]]: Mapping of URI to images
"""
...Usage example:
class MyLibraryProvider(LibraryProvider):
def search(self, query=None, uris=None, exact=False):
# Implement search logic for your music source
tracks = self._search_tracks(query)
artists = self._search_artists(query)
albums = self._search_albums(query)
return SearchResult(
uri=f"myservice:search:{query}",
tracks=tuple(tracks),
artists=tuple(artists),
albums=tuple(albums)
)Playback Provider Interface
Provider interface for controlling audio playback including URI translation and stream preparation.
class PlaybackProvider:
"""Base class for playback providers handling audio stream control."""
def __init__(self, audio, backend): ...
def translate_uri(self, uri):
"""
Translate Mopidy URI to playable URI.
Parameters:
- uri (str): Mopidy track URI
Returns:
- str or None: Playable URI for audio system
"""
...
def get_time_position(self):
"""
Get current playback position.
Returns:
- int: Position in milliseconds
"""
...
def pause(self):
"""
Pause playback.
Returns:
- bool: True if successful
"""
...
def resume(self):
"""
Resume playback.
Returns:
- bool: True if successful
"""
...
def seek(self, time_position):
"""
Seek to position.
Parameters:
- time_position (int): Position in milliseconds
Returns:
- bool: True if successful
"""
...
def stop(self):
"""
Stop playback.
Returns:
- bool: True if successful
"""
...
def prepare_change(self):
"""
Indicate that a URI change is about to happen.
Note: This is primarily for internal use between backends and core.
Returns:
- None
"""
...
def change_track(self, track):
"""
Switch to provided track.
Note: This is primarily for internal use between backends and core.
Parameters:
- track (Track): Track to switch to
Returns:
- bool: True if successful
"""
...
def is_live(self, uri):
"""
Decide if URI should be treated as a live stream.
Playing as live stream disables buffering and reduces latency.
Parameters:
- uri (str): URI to check
Returns:
- bool: True if URI is a live stream
"""
...
def should_download(self, uri):
"""
Decide if URI should use progressive download buffering.
For fixed-length files, buffering the entire file improves playback.
Parameters:
- uri (str): URI to check
Returns:
- bool: True if should use download buffering
"""
...
def on_source_setup(self, source):
"""
Called when a new GStreamer source is created for configuration.
This runs in the audio thread and should not block.
Parameters:
- source (GstElement): GStreamer source element
Returns:
- None
"""
...Usage example:
class MyPlaybackProvider(PlaybackProvider):
def translate_uri(self, uri):
if uri.startswith("myservice:"):
# Convert to streamable URL
track_id = uri.split(":")[-1]
return f"https://api.myservice.com/stream/{track_id}"
return NonePlaylists Provider Interface
Provider interface for managing playlists including creation, modification, and deletion.
class PlaylistsProvider:
"""Base class for playlist providers handling playlist management."""
def __init__(self, backend): ...
def as_list(self):
"""
Get all playlists.
Returns:
- list[Playlist]: Available playlists
"""
...
def get_items(self, uri):
"""
Get playlist contents.
Parameters:
- uri (str): Playlist URI
Returns:
- list[Ref]: Playlist items
"""
...
def create(self, name):
"""
Create new playlist.
Parameters:
- name (str): Playlist name
Returns:
- Playlist or None: Created playlist
"""
...
def delete(self, uri):
"""
Delete playlist.
Parameters:
- uri (str): Playlist URI to delete
Returns:
- bool: True if successful
"""
...
def lookup(self, uri):
"""
Lookup playlist by URI.
Parameters:
- uri (str): Playlist URI
Returns:
- Playlist or None: Found playlist
"""
...
def refresh(self):
"""
Refresh playlists cache.
Returns:
- bool: True if successful
"""
...
def save(self, playlist):
"""
Save playlist changes.
Parameters:
- playlist (Playlist): Playlist to save
Returns:
- Playlist or None: Saved playlist
"""
...Usage example:
class MyPlaylistsProvider(PlaylistsProvider):
def create(self, name):
# Create playlist in your service
playlist_id = self._api_create_playlist(name)
return Playlist(
uri=f"myservice:playlist:{playlist_id}",
name=name,
tracks=()
)
def save(self, playlist):
# Save playlist changes to your service
playlist_id = playlist.uri.split(":")[-1]
track_uris = [track.uri for track in playlist.tracks]
self._api_update_playlist(playlist_id, track_uris)
return playlistBackend Event Listeners
Event notification interfaces for backend state changes and lifecycle events.
class BackendListener:
"""Base class for backend event listeners."""
def playlists_loaded(self):
"""Called when backend playlists are loaded."""
...
class LibraryListener:
"""Base class for library event listeners."""
def library_changed(self, uri):
"""
Called when library content changes.
Parameters:
- uri (str): URI that changed
"""
...
class PlaybackListener:
"""Base class for playback event listeners."""
def reached_end_of_stream(self):
"""Called when stream reaches end."""
...
def position_changed(self, position):
"""
Called when playback position changes.
Parameters:
- position (int): New position in milliseconds
"""
...
def state_changed(self, old_state, new_state):
"""
Called when playback state changes.
Parameters:
- old_state (str): Previous state
- new_state (str): New state
"""
...URI Scheme Management
Utilities for working with URI schemes and backend registration.
def get_backend_names():
"""
Get names of all available backends.
Returns:
- list[str]: Backend names
"""
...
def get_backends_by_uri_scheme(uri_scheme):
"""
Get backends that handle a specific URI scheme.
Parameters:
- uri_scheme (str): URI scheme to match
Returns:
- list[Backend]: Matching backends
"""
...Implementation Guidelines
Backend Registration
Backends are typically registered through Mopidy's extension system:
from mopidy.ext import Extension
class MyExtension(Extension):
def setup(self, registry):
from .backend import MyBackend
registry.add('backend', MyBackend)Error Handling
Backends should handle errors gracefully and return appropriate fallback values:
class MyLibraryProvider(LibraryProvider):
def search(self, query=None, uris=None, exact=False):
try:
return self._perform_search(query)
except ConnectionError:
logger.warning("Search failed due to connection error")
return SearchResult(uri="", tracks=(), artists=(), albums=())URI Format Conventions
Use consistent URI formats for your backend:
# Examples of good URI formats:
# myservice:track:123456
# myservice:album:789012
# myservice:artist:345678
# myservice:playlist:901234Caching Considerations
Implement appropriate caching for improved performance:
class MyLibraryProvider(LibraryProvider):
def __init__(self, backend):
super().__init__(backend)
self._cache = {}
self._cache_timeout = 300 # 5 minutes
def lookup(self, uris):
results = {}
uncached_uris = []
for uri in uris:
if uri in self._cache:
results[uri] = self._cache[uri]
else:
uncached_uris.append(uri)
if uncached_uris:
fresh_results = self._api_lookup(uncached_uris)
self._cache.update(fresh_results)
results.update(fresh_results)
return results