plistsync.core.collection

Collection Protocols for Track Management.

This module defines a set of protocols that model different capabilities a track collection might support. These protocols use Python’s structural subtyping (PEP 544) rather than inheritance-based interfaces, offering flexibility in composing complex behaviors while maintaining strong type safety and clarity.

Key Design Principles:

1. Capability-based Design: Collections declare what operations they support by implementing specific protocols:

   • GlobalLookup: Enables exact matching via globally unique identifiers.

   • LocalLookup: Supports context-specific identifier matching.

   • InfoLookup: Facilitates metadata-based similarity searches.

   • TrackStream: Provides iteration and bulk processing abilities.

2. Progressive Enhancement: Collections can implement additional protocols for more sophisticated matching strategies, all while maintaining backward compatibility with basic iteration.

3. Runtime Flexibility: The @runtime_checkable decorator allows collections to be verified at runtime, while static type checkers can verify protocol compliance during development.

The main Collection abstract base class (ABC) demonstrates the integration of these protocols into a comprehensive track matching strategy via the match method. Developers are encouraged to extend the Collection class to create new collection types with different internal storage strategies (e.g. in-memory, databases).

Usage Example:

Create a custom collection by implementing the desired protocols and extend the Collection ABC, ensuring that the match method efficiently leverages all relevant capabilities offered by the collection.

class MyTrackCollection(Collection, GlobalLookup, LocalLookup, TrackStream):
    # Implement required methods...

Overview

Classes

GlobalLookup	A collection that can find tracks using global unique IDs.
LocalLookup	A collection that can find tracks using local context-specific IDs.
InfoLookup	A collection that can search for tracks using metadata.
TrackStream	Supports iteration and parallel processing of tracks.
Collection	A generic data structure that allows lookup or iteration of tracks.
Library	Represents a collection of tracks in a library with playlist management.

Module Contents

class plistsync.core.collection.GlobalLookup

Bases: Protocol, Generic[T]

A collection that can find tracks using global unique IDs.

abstractmethod find_by_global_ids(global_ids: plistsync.core.track.GlobalTrackIDs) → T | None

Find a single track by its global identifiers.

find_many_by_global_ids(global_ids_list: collections.abc.Iterable[plistsync.core.track.GlobalTrackIDs]) → collections.abc.Iterable[T | None]

Find multiple tracks by their global identifiers.

Default implementation iterates over the provided list and calls find_by_global_ids for each entry. Collections can override this method to provide a more efficient batch lookup if supported.

class plistsync.core.collection.LocalLookup

Bases: Protocol, Generic[T]

A collection that can find tracks using local context-specific IDs.

abstractmethod find_by_local_ids(local_ids: plistsync.core.track.LocalTrackIDs) → T | None

Find a single track by its local identifiers.

find_many_by_local_ids(local_ids_list: collections.abc.Iterable[plistsync.core.track.LocalTrackIDs]) → collections.abc.Iterable[T | None]

Find multiple tracks by their local identifiers.

Default implementation iterates over the provided list and calls find_by_local_ids for each entry. Collections can override this method to provide a more efficient batch lookup if supported.

class plistsync.core.collection.InfoLookup

Bases: Protocol, Generic[T]

A collection that can search for tracks using metadata.

abstractmethod find_by_info(info: plistsync.core.track.TrackInfo) → collections.abc.Iterable[T]

Find tracks matching the given metadata.

class plistsync.core.collection.TrackStream

Bases: Protocol, Generic[T]

Supports iteration and parallel processing of tracks.

A collection implementing this protocol must support iteration, yielding Track objects one by one. This makes it possible to traverse all tracks in the collection, for example when scanning a library or processing all items in a playlist.

map_threadpool_chunked(func: collections.abc.Callable[Concatenate[T, P], R], chunk_size: int = 100, max_workers: int = 4, *args: P, **kwargs: P) → collections.abc.Iterable[tuple[collections.abc.Sequence[R], collections.abc.Sequence[T]]]

Map a function to each track in parallel.

Iterate over all tracks in the collection and apply a function to each track. Use a threadpool to parallelize a computation. This method should be used to parallelize compute heavy operations on the collection or to speed up the processing of large collections.

To allow processing large collections we process the collection in chunks of chunk_size tracks. This should help to reduce the memory footprint.

Parameters:

• func (Callable[[Track], T]) – The function to apply to each element in the collection. First argument should be a track.

• chunk_size (int) – The maximum number of tracks to process in each chunk.

• **kwargs (Any) – Additional keyword arguments to passed to each function call.

Example

If you want to apply a function to each track in the collection, you can use this method like this:

def heavy_computation(track: Track, *args) -> int:
    pass # Do some heavy computation on the track and return a result

for results, tracks in collection.map_threadpool(
    heavy_computation,
    chunk_size=100,
    *args,
):
    # do something with the results and related tracks
    pass

map_threadpool(func: collections.abc.Callable[Concatenate[T, P], R], chunk_size: int = 100, max_workers: int = 4, *args: P, **kwargs: P) → collections.abc.Iterable[tuple[R, T]]

Map a function to each track in parallel and return a list of results.

This is a convenience method that uses map_threadpool_chunked to process the entire collection and return a flat list of results.

class plistsync.core.collection.Collection

Bases: abc.ABC, Generic[T]

A generic data structure that allows lookup or iteration of tracks.

Collections act as flexible track containers, accommodating multiple storage formats and sources, such as online databases or local files, without dictating a specific storage mechanism.

This abstract base class is designed to support adaptable implementations for accessing and interacting with tracks in diverse ways, see protocols above.

match(track: plistsync.core.track.Track, skip_after_local_match: bool = True, skip_after_perfect_fuzzy_match: bool = True, cutoff=0.6) → plistsync.core.matching.Matches[T]

Potential matches for the given track based on different lookup strategies.

The method checks for matches in this order:

1. Global IDs (exact match, returns immediately if found)

2. Local IDs (exact match with similarity check)

3. Track info (similarity-based search)

4. Fallback to iterating through all tracks if needed. This still uses the three methods above, but is way less efficient.

Parameters:

• track – The track to match against this collection

• skip_after_local_match – If True, return after first successful match when searching local IDs

• skip_after_perfect_fuzzy_match – If True, return after finding a perfect fuzzy match (similarity == 1.0)

• cutoff – Minimum similarity score (0-1) for a match to be considered

class plistsync.core.collection.Library

Bases: Generic[T, Plist], Collection[T]

Represents a collection of tracks in a library with playlist management.

This class serves as a base for library collections across diverse services. It provides a framework for managing tracks and playlists, allowing each service to implement its specifics.

property name: str

Name of the library, typically the service name.

property playlists: collections.abc.Iterable[Plist]

Abstractmethod:

Retrieve playlists associated with this library collection.

abstractmethod get_playlist(*, id: plistsync.core.playlist.PlaylistID | str | None = None) → Plist | None

Get a playlist by identifier.

Implement with kwargs like name=, ids=, url=, or uri=. Return None for searches that fail.

abstractmethod create_playlist(name: str, description: str | None = None, tracks: list[T] | None = None) → Plist

Create a new playlist.

get_playlist_or_raise(*, id: plistsync.core.playlist.PlaylistID | str | None = None, **kwargs) → Plist

Like get_playlist() but raises if no result is found.