"""TidalClient — the public entry point of tidalpy.
One :class:`TidalClient` owns its own HTTP session, OAuth token
lifecycle, and per-instance caches: two clients in one process never
share credentials or state (Django multi-worker / Celery prefork
safe). The client is a *facade* over the working parts:
- :class:`~tidalpy.transport.Transport` — HTTP with Tidal's policy;
- :class:`~tidalpy.oauth.OAuth` — device flow, refresh, token store;
- :class:`~tidalpy.api.TidalApi` — catalog, search, stream resolution;
- :class:`~tidalpy.pipeline.Pipeline` — collection downloads;
- :class:`~tidalpy.streaming.MediaStream` — bytes to the caller, no disk.
Typical Django/Celery usage::
client = TidalClient(Config(token_path=..., output=...))
if not client.auth_status()["logged_in"]:
auth = client.start_device_login()
# show auth.verification_url to an operator, then:
client.finish_device_login(auth, timeout=300)
report = client.download(url, quality=Quality.LOSSLESS,
on_progress=lambda e: update_task(e))
for item in report.failures:
log(item.to_dict())
"""
from __future__ import annotations
import logging
import re
import threading
from collections.abc import Iterator, Sequence
from pathlib import Path
from typing import Any, Callable
from .api import TidalApi, artwork_url, parse_tidal_url
from .config import Config
from .errors import TidalError
from .models import (
Album,
Artist,
DownloadReport,
ItemReport,
Playlist,
Resource,
Track,
)
from .oauth import DeviceAuth, OAuth
from .pipeline import Pipeline, ProgressCallback
from .quality import Quality, parse_quality
from .session import FileTokenStore, TokenRecord, TokenStore
from .streaming import DEFAULT_CHUNK_SIZE, MediaStream
from .transport import Transport
logger = logging.getLogger("tidalpy.client")
def _normalize_artist_name(text: Any) -> str:
"""Case/punctuation-insensitive comparison key for artist names.
The catalog spells the same artist several ways ("O.S.T.R." vs
"OSTR" on different rows); only the alphanumerics are load-bearing
for the "is this query that artist's name" check.
"""
return re.sub(r"[^a-z0-9]+", "", str(text or "").casefold())
def _album_search_row(album: Album) -> dict[str, Any]:
"""The :meth:`TidalClient.search` album-row shape for an ``Album``.
Discography entries are catalog ``Album`` objects, not search rows;
they must wear the same shape so a host can render a
:meth:`TidalClient.search_albums` result exactly like an album
search page.
"""
row: dict[str, Any] = {
"kind": "album",
"id": album.id,
"title": album.title or "",
"artist": ", ".join(a.name for a in album.artists if a.name),
}
if album.version and album.version.strip():
row["version"] = album.version
if album.audio_quality:
row["quality"] = album.audio_quality
if isinstance(album.artwork_id, str) and album.artwork_id:
row["artwork"] = album.artwork_id
return row
def _failure_report(
kind: str,
title: str | None,
item_id: Any,
reason: str,
exc: TidalError,
requested_quality: str,
) -> DownloadReport:
"""A single-item failed report for a URL the pipeline never
processed (resolve or run-level failure). Keeps
:meth:`TidalClient.download_many`'s contract — one report per URL,
with the machine ``code``/``retryable`` a host can key off."""
report = DownloadReport(
kind=kind,
title=title,
requested_quality=requested_quality,
)
report.items.append(
ItemReport(
index=1,
kind=kind,
id=item_id,
title=title,
status="failed",
reason=reason,
code=exc.code,
retryable=exc.retryable,
)
)
return report
[docs]
class TidalClient:
"""Main interface for Tidal operations (catalog, auth, download)."""
[docs]
def __init__(
self,
config: Config | None = None,
*,
token_store: TokenStore | None = None,
media_resolver: Callable[[str], tuple[str, ...]] | None = None,
clock: Callable[[], float] | None = None,
sleeper: Callable[[float], None] | None = None,
) -> None:
"""Create a client.
Args:
config: A :class:`~tidalpy.config.Config` (defaults are safe).
token_store: Where the saved login lives. Defaults to a
:class:`~tidalpy.session.FileTokenStore` at
``config.token_file``; Django may pass a DB-backed store.
media_resolver: Override for media-URL SSRF validation;
intended as a test hook (e.g. a local test server).
Production code should leave this ``None``.
clock: Monotonic clock override (tests).
sleeper: Sleep override for the device-login poll (tests);
expected to be interruptible.
"""
self.config = config or Config()
self._transport = Transport(
self.config, stop_event=None, media_resolver=media_resolver
)
store = token_store or FileTokenStore(self.config.token_file)
self._oauth = OAuth(
self.config, self._transport, store, clock=clock, sleeper=sleeper
)
self._api = TidalApi(self.config, self._oauth, self._transport)
self._pipeline = Pipeline(
self._api, self.config, self._transport, media_resolver=media_resolver
)
self._lock = threading.Lock()
self._active_stops: list[threading.Event] = []
self._is_closed = False
# -- lifecycle -----------------------------------------------------------
def __enter__(self) -> "TidalClient":
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
[docs]
def close(self) -> None:
"""Close the HTTP session (idempotent)."""
if not self._is_closed:
self._transport.close()
self._is_closed = True
# -- authentication --------------------------------------------------------
[docs]
def auth_status(self) -> dict[str, Any]:
"""Current login state (plus subscription facts when logged in).
Verifies the login the way a download would use it: the
subscription call goes through the transport's 401 handshake,
so an expired access token with a valid refresh token is
refreshed *before* the status is reported — the host never sees
``expired: True`` for a login that the next request would
transparently repair.
Returns a JSON-safe dict: ``logged_in``, ``userId``,
``countryCode``, ``clientId``, ``tokenType``, ``expired``,
``secondsLeft``; ``subscription`` is added when logged in.
"""
status = self._oauth.status()
if status.get("logged_in"):
subscription = self._api.subscription()
status = self._oauth.status() # post-refresh state
status["subscription"] = subscription
return status
@property
def logged_in(self) -> bool:
value = self.auth_status().get("logged_in", False)
return bool(value)
def __repr__(self) -> str:
"""A compact shell/log line — local state only (never
triggers the network the way :meth:`logged_in` does)."""
record = self._oauth.record
if record is None or not record.present:
auth = "logged-out"
elif record.expired:
auth = f"user {record.user_id}, token expired"
else:
auth = f"user {record.user_id}"
return (
f"TidalClient(output={self.config.output!r}, "
f"quality={self.config.quality.value!r}, {auth})"
)
[docs]
def login(self, access_token: str, user_id: int | str | None = None) -> TokenRecord:
"""Adopt a token the operator pasted in (browser session).
Probes the token against Tidal's ``sessions`` endpoint before
persisting it; raises :class:`~tidalpy.errors.AuthError` when
rejected.
"""
return self._oauth.login_with_token(access_token, user_id=user_id)
[docs]
def start_device_login(self) -> DeviceAuth:
"""Begin the device flow; show ``verification_url`` to a human."""
return self._oauth.start_device_login()
[docs]
def finish_device_login(
self,
auth: DeviceAuth,
timeout: float | None = None,
user_id: int | str | None = None,
) -> TokenRecord:
"""Poll until the operator approves (or the deadline passes)."""
return self._oauth.finish_device_login(auth, timeout=timeout, user_id=user_id)
[docs]
def logout(self) -> bool:
"""End the session: best-effort remote revocation (Tidal's
``/v1/logout``), then forget the saved session locally.
A failed revocation never blocks the logout — the local session
is cleared either way. Returns whether the remote revocation
succeeded (``False`` when logged out already or the request
failed; the library logs the outcome).
"""
return self._oauth.logout()
# -- cancellation ----------------------------------------------------------
[docs]
def cancel(self) -> None:
"""Cancel every in-flight operation on this client.
Cancellation is cooperative: current requests finish their chunk
and the next check raises
:class:`~tidalpy.errors.CancelledError`.
"""
with self._lock:
events = list(self._active_stops)
for event in events:
event.set()
@property
def cancelled(self) -> bool:
"""True when a cancellation was requested for an active operation."""
with self._lock:
return any(event.is_set() for event in self._active_stops)
def _begin_operation(
self, cancel_event: threading.Event | None
) -> tuple[threading.Event, bool]:
"""Register the stop event for one operation (per-operation)."""
event = cancel_event or threading.Event()
with self._lock:
self._active_stops.append(event)
self._transport.set_stop_event(event)
return event, cancel_event is not None
def _end_operation(self, event: threading.Event, owned: bool) -> None:
with self._lock:
try:
self._active_stops.remove(event)
except ValueError:
pass
if owned:
self._transport.set_stop_event(threading.Event())
# -- catalog ---------------------------------------------------------------
[docs]
def resolve(self, url: str) -> Resource:
"""Resolve any Tidal link (or bare id) to a catalog resource.
Raises:
NotFoundError: unknown id or empty URL.
UnsupportedType: video links (audio-only library).
"""
return self._api.resolve(url)
[docs]
def resolve_many(
self,
urls: Sequence[str],
cancel_event: threading.Event | None = None,
) -> list[dict[str, Any]]:
"""Resolve many links (or bare ids) with per-item isolation —
the indexer's pre-flight before spending a download on the
batch.
A pasted/exports blob (``parse_link_list``) carries the odd
dead id, a video link, or a typo; the walk must not die on one
bad row — the same isolation contract :meth:`download_many`
applies to downloads. Each row is JSON-safe, one per URL in
input order:
- on success: ``{"url", "ok": True, "kind", "id", "title",
"link"}`` (``kind`` the lowercase resource kind, ``id`` the
linkable id, ``link`` the canonical URL);
- on failure: ``{"url", "ok": False, "code", "reason"}``
(the machine ``code`` for library errors — ``"not-found"``,
``"unsupported-type"``, … — the exception's name otherwise).
The walk is sequential (resolutions are catalog-light;
parallel calls would only fight over the rate-limit budget).
``cancel_event`` ends the walk: the remaining URLs are not
attempted, and the returned list covers the URLs processed.
Args:
urls: Tidal links (or bare ids), in walk order.
cancel_event: Optional cancellation (see
:meth:`download_many`).
Returns:
One row per processed URL, in input order.
"""
rows: list[dict[str, Any]] = []
for url in urls:
if cancel_event is not None and cancel_event.is_set():
break
row: dict[str, Any] = {"url": url, "ok": False}
try:
resource = self.resolve(url)
except TidalError as exc:
row["code"] = exc.code
row["reason"] = str(exc)
except Exception as exc: # noqa: BLE001 - isolation contract
row["code"] = type(exc).__name__
row["reason"] = str(exc)
else:
row.update(
ok=True,
kind=resource.kind,
id=resource.data.id,
title=resource.title,
link=resource.link,
)
rows.append(row)
return rows
[docs]
def artwork(self, item: str, size: int = 1280) -> bytes | None:
"""The cover art for a link, row artwork id, or image URL.
Accepts, in order of preference:
- a five-segment artwork id (the ``artwork`` value search rows
and ``metadata()`` dicts carry);
- a ``resources.tidal.com/images/...`` URL (the id is parsed
out of the path — a host can hand back what
:meth:`artwork_url` built);
- any Tidal link (or bare id): it is resolved and the artwork
id of the resource (album cover, playlist cover, artist
picture, track's album cover) is used.
The fetch goes through the SSRF-validated transport (the same
validated-redirect-follow the download pipeline uses), so a
host can serve these bytes without doing its own HTTP. Best
effort: a failed fetch (no artwork, dead CDN, refused host)
returns ``None`` and never raises — cover art is garnish.
Args:
item: Artwork id, image URL, or Tidal link / bare id.
size: Square dimension in pixels, 1..8192.
Returns:
The image bytes (JPEG), or ``None`` when unavailable.
"""
artwork_id = self._artwork_id_of(item)
if not artwork_id:
return None
# Caller errors (a bad size) raise; only the *fetch* is best
# effort — cover art is garnish.
url = artwork_url(artwork_id, size, size)
try:
return self._transport.get_bytes(url)
except Exception: # noqa: BLE001 - cover art is garnish, never fatal
logger.debug("artwork fetch failed for %s", artwork_id, exc_info=True)
return None
def _artwork_id_of(self, item: str) -> str | None:
"""Normalize an :meth:`artwork` argument to an artwork id."""
text = (item or "").strip()
if not text:
return None
if re.fullmatch(
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}"
r"-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}",
text,
):
return text
match = re.search(
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}"
r"-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}",
text.replace("/", "-"),
)
if match:
return match.group(0)
# A Tidal link (or bare id): resolve and take the artwork.
try:
resource = self._api.resolve(item)
except TidalError:
return None
data = resource.data
if isinstance(data, Track):
return data.album.artwork_id if data.album is not None else None
return getattr(data, "artwork_id", None) or getattr(data, "picture", None)
[docs]
def artist_albums(
self, url: str, include_ep: bool | None = None
) -> list[dict[str, Any]]:
"""The discography of an artist link (or bare artist id).
Args:
url: ``https://tidal.com/artist/<id>`` or a bare id.
include_ep: also include EPs and singles (a second,
filtered listing). Defaults to
:data:`~tidalpy.config.Config.include_ep` (on, matching
the reference client's effective default).
Returns:
One :class:`~tidalpy.models.Album` dict per album (id,
title, release_date, …).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
artist = resource.data
if resource.kind != "artist" or not isinstance(artist, Artist):
raise TidalError(
"artist_albums() requires an artist link; "
f"got a {resource.kind} link"
)
if artist.id is None:
raise TidalError("resolved artist has no id")
if include_ep is None:
include_ep = self.config.include_ep
return [
album.to_dict()
for album in self._api.get_artist_albums(artist.id, include_ep=include_ep)
]
[docs]
def artist_top_tracks(
self,
url: str,
limit: int = 20,
) -> list[Track]:
"""The artist's top tracks, ranked by the catalog.
The v1 ``artists/{id}/toptracks`` endpoint (live API verified:
paged, full track payloads — the reference clients don't
expose it). A real-time "top tracks" tab for an artist page:
the tracks come back ranked, as :class:`~tidalpy.models.Track`
models, each with its album back-reference set by the catalog
payload.
Args:
url: An artist link (or bare artist id).
limit: 1..50 tracks (the top N — the endpoint is ranked).
Returns:
The ranked tracks (fewer when the catalog has fewer).
Raises:
AuthError: not logged in.
TidalError: resolution failures (bad URL, unknown artist).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
artist = resource.data
if resource.kind != "artist" or not isinstance(artist, Artist):
raise TidalError(
"artist_top_tracks() requires an artist link; "
f"got a {resource.kind} link"
)
if artist.id is None:
raise TidalError("resolved artist has no id")
return self._api.get_artist_top_tracks(artist.id, limit=limit)
[docs]
def tracks(self, url: str) -> list[Track]:
"""The link's tracks as :class:`~tidalpy.models.Track` models,
in catalog order.
The content-listing facade for real-time UIs and indexers: an
album page (title, number, duration, explicit, bpm/key per
row), a playlist page, an artist's flattened discography
(each track carrying its ``album`` back-reference), or a
single track (one-element list). The listing is the catalog's
— no stream is resolved, so it is cheap and rate-limit-light
(one items call per collection, cached per client instance).
Non-audio slots (music videos) are skipped; video links are
not supported by this library and raise as in
:meth:`resolve`.
Args:
url: Any Tidal link (or bare id) this library resolves:
album, playlist, mix, artist, or single track.
Returns:
The tracks in catalog order (album: track number;
playlist/mix: playlist position; artist: discography
order, per-album).
Raises:
AuthError: not logged in.
TidalError: resolution failures (bad URL, unknown
resource); video links (UnsupportedType).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
return [track for track in self._api.get_items(resource) if track is not None]
[docs]
def tracks_by_ids(self, track_ids: Sequence[int | str]) -> list[Track]:
"""Bulk track metadata (the indexer facade).
The ids-first counterpart of :meth:`tracks`: a host that has
track ids from elsewhere (a playlist export, a library audit,
a web-app deep link) gets the full Track models in one
catalog round trip per 50 ids, instead of N single-track
resolves — each of which starts as a rate-limited fetch.
Result order is the request order (duplicates collapse to the
first occurrence); ids the catalog does not know are skipped,
not fabricated — diff the result against the request to find
the gaps.
Args:
track_ids: Track ids (legacy numeric or web-app ids).
Returns:
The Track models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_tracks_by_ids(track_ids)
[docs]
def track_by_isrc(self, isrc: str) -> Track | None:
"""Resolve a track by its ISRC (the indexer facade).
The key-based counterpart of :meth:`tracks_by_ids`: a host
that holds ISRCs from elsewhere (CDDB, a licensing feed,
another catalog's export) instead of Tidal ids gets the full
Track model in one catalog call. A lookup, not a fetch: an
ISRC the catalog does not know returns ``None``, so a
library diff is a plain loop with no exception per miss.
The input is normalised (whitespace removed, upper-cased —
the catalog rejects any other case); blank input returns
``None`` without a request.
Args:
isrc: The track's ISRC (12 characters, e.g.
``USUM71801211``).
Returns:
The Track model, or ``None`` when the catalog has no
track with that ISRC.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures (a malformed ISRC is the
catalog's 400, which propagates).
"""
self._ensure_logged_in()
return self._api.get_track_by_isrc(isrc)
[docs]
def tracks_by_isrcs(self, isrcs: Sequence[str]) -> list[Track]:
"""Bulk track metadata by ISRC (the indexer facade).
The bulk sibling of :meth:`track_by_isrc`: a host that holds
a library of ISRCs from elsewhere (a CDDB export, a
licensing feed) instead of Tidal ids gets the full Track
models in one catalog round trip per 50 ISRCs, instead of N
single lookups. Same contract as :meth:`tracks_by_ids`:
request order, duplicates collapse, ISRCs the catalog does
not know are skipped, not fabricated (diff to find the
gaps).
Each ISRC is normalised (whitespace removed, upper-cased);
blank entries are skipped.
Args:
isrcs: Track ISRCs (12 characters each, e.g.
``USUM71801211``).
Returns:
The Track models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_tracks_by_isrcs(isrcs)
[docs]
def albums_by_ids(self, album_ids: Sequence[int | str]) -> list[Album]:
"""Bulk album metadata (the indexer facade).
The album counterpart of :meth:`tracks_by_ids` (an indexer
importing a discography from ids): full Album models in one
catalog round trip per 50 ids. Same contract — request
order, duplicates collapse, unknown ids skipped not
fabricated (diff to find the gaps).
The v2 bulk rows carry no cover art (the single-album
resource does), so bulk albums have no artwork.
Args:
album_ids: Album ids (legacy numeric or web-app ids).
Returns:
The Album models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_albums_by_ids(album_ids)
[docs]
def album_by_upc(self, upc: str) -> Album | None:
"""Resolve an album by its UPC (the indexer facade).
The album counterpart of :meth:`track_by_isrc`: a host that
holds UPCs from elsewhere (the back of a CD case, a licensing
feed) instead of Tidal ids gets the full Album model in one
catalog call. A lookup, not a fetch: a UPC the catalog does
not know returns ``None``, so a library diff is a plain loop
with no exception per miss.
The input is stripped and otherwise passed through (UPCs run
12-14 digits, no reformatting); blank input returns ``None``
without a request.
Args:
upc: The album's UPC / barcode (e.g.
``634904078164``).
Returns:
The Album model, or ``None`` when the catalog has no
album with that barcode.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures (a malformed value is the
catalog's 400, which propagates).
"""
self._ensure_logged_in()
return self._api.get_album_by_upc(upc)
[docs]
def albums_by_upcs(self, upcs: Sequence[str]) -> list[Album]:
"""Bulk album metadata by UPC (the indexer facade).
The bulk sibling of :meth:`album_by_upc`: a host that holds
a library of barcodes from elsewhere (a shelf of CD cases, a
licensing feed) instead of Tidal ids gets the full Album
models in one catalog round trip per 50 UPCs, instead of N
single lookups. Same contract as :meth:`albums_by_ids`:
request order, duplicates collapse, barcodes the catalog
does not know are skipped, not fabricated (diff to find the
gaps).
Each UPC is stripped and otherwise passed through (no
reformatting); blank entries are skipped.
Args:
upcs: Album UPCs / barcodes (e.g.
``634904078164``).
Returns:
The Album models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_albums_by_upcs(upcs)
[docs]
def artists_by_ids(self, artist_ids: Sequence[int | str]) -> list[Artist]:
"""Bulk artist metadata (the indexer facade).
The artist counterpart of :meth:`tracks_by_ids` — also the
only v2 artist resolution path (the single artist fetch is
v1), so web-app ids the v1 catalog does not know are
resolvable here. Same contract: request order, duplicates
collapse, unknown ids skipped not fabricated (diff to find
the gaps).
The v2 bulk rows carry no picture/handle, so those stay
``None`` on bulk artists.
Args:
artist_ids: Artist ids (legacy numeric or web-app ids).
Returns:
The Artist models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_artists_by_ids(artist_ids)
[docs]
def playlists_by_ids(self, playlist_ids: Sequence[int | str]) -> list[Playlist]:
"""Bulk playlist metadata (the indexer facade).
The playlist counterpart of :meth:`tracks_by_ids`: a host
that has playlist UUIDs from elsewhere (a web-app export, a
cross-device library sync) gets the full Playlist models in
one catalog round trip per 50 UUIDs, instead of N resolves.
Same contract: request order, duplicates collapse, unknown
UUIDs skipped not fabricated (diff to find the gaps).
The rows carry the card facts (title, description, track
count, duration, follower count) but no artwork: a host
that needs a cover resolves that playlist singly.
Args:
playlist_ids: Playlist UUIDs (the linkable form).
Returns:
The Playlist models in request order.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_playlists_by_ids(playlist_ids)
[docs]
def artwork_url(
self, artwork_id: str, width: int = 1280, height: int = 1280
) -> str:
"""The static-CDN image URL for a five-segment artwork id
(the ``artwork`` value of search rows, ``Album.artwork_id``).
Pure URL construction — no network, no auth: search and
metadata clients render thumbnails straight from the ids the
catalog returns. See :func:`tidalpy.artwork_url`.
"""
return artwork_url(artwork_id, width, height)
[docs]
def lyrics(self, url: str) -> dict[str, str]:
"""The lyrics of a track link, ``{"lyrics", "subtitles"}``.
``lyrics`` is the plain text, ``subtitles`` the timed (LRC)
version; each is ``""`` when Tidal has none (lyrics never
error — a track without them simply returns both empty).
Requires a single track link.
When Tidal has plain lyrics but no timed subtitles for this
exact version (a remaster, a live edition), a fallback
searches the catalog for a matching version and adopts its
timed subtitles (keeping this track's plain text) — so
display clients get synced lyrics for the versions Tidal
tags up as well.
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
track = resource.data
if resource.kind != "track" or not isinstance(track, Track) or track.id is None:
raise TidalError(
"lyrics() requires a single track; " f"got a {resource.kind} link"
)
# The resolved track lets the timed-lyrics fallback run (when
# Tidal has no subtitles for this exact version, a matching
# remaster/live version may still carry timed lyrics).
return self._api.get_lyrics(track.id, track)
[docs]
def track_contributors(self, url: str) -> list[dict[str, Any]]:
"""The credits of a track link (the track page's "credits" tab).
Each row is ``{"name", "role"}`` (the catalog's credit
strings: Producer, Composer, Lyricist, Associated Performer,
Music Publisher, ...). A track with no credits returns ``[]``
(a legitimate empty credit list, not an error); a link that
is not a resolvable track raises like the other track
facades. Requires a single track link (or bare track id).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
track = resource.data
if resource.kind != "track" or not isinstance(track, Track) or track.id is None:
raise TidalError(
"track_contributors() requires a single track; "
f"got a {resource.kind} link"
)
return self._api.get_track_contributors(track.id)
[docs]
def search(
self,
query: str,
item_type: str | None = None,
limit: int = 20,
offset: int = 0,
country: str | None = None,
) -> dict[str, Any]:
"""Search the catalog; returns the reference client's search shape.
Result: ``{"text", "type", "total", "items"}`` where each item
is ``{"kind", "id", "title"}`` plus ``artist`` for albums and
tracks (the same rows the legacy ``search`` RPC/CLI emit — the
``id`` values build ``https://tidal.com/<kind>/<id>`` links
straight away; playlist ids are the ``uuid``). Rows
additionally carry the catalog's garnish *when present*:
``version`` (track/album editions), ``duration`` in seconds
(track/video length, album total), ``quality`` (track/album,
the raw API value). ``total`` is the catalog's per-group sum,
so it may exceed the number of rows returned.
Args:
query: Search text.
item_type: ``"track"`` / ``"album"`` / ``"artist"`` /
``"playlist"`` / ``"video"``; ``None`` (default)
searches all five groups (reference-client parity).
The result's ``type`` is the matching reference-client
``Type`` name (``"Null"`` for the default all-groups
search).
limit: 1..50 results per group.
offset: Result offset for paging through the catalog.
country: Search this market's catalog (ISO country code,
e.g. ``"US"``) instead of the account's — a search
UI serving an account in another region. ``None``
(default) uses the session's country.
"""
types = (item_type.upper(),) if item_type else None
items, total, top_hit = self._api.search(
query,
types=types or ("ARTIST", "ALBUM", "TRACK", "VIDEO", "PLAYLIST"),
limit=limit,
offset=offset,
country=country,
)
# The reference client's Type name for the requested group
# ("tracks" -> "Track", as its SEARCH_TYPES table maps it;
# Type.Null -> "Null" on the default all-groups search).
type_name = item_type.strip().rstrip("s").capitalize() if item_type else "Null"
wrapper: dict[str, Any] = {
"text": query,
"type": type_name,
"total": total,
"items": items,
}
if top_hit is not None:
# The catalog's single best match for the query (its
# topHit field — the reference clients ignore it): the
# "best match" banner of a real-time search UI, wearing
# the same row shape as the results. Present only when
# the catalog names one (garnish convention).
wrapper["top_hit"] = top_hit
return wrapper
[docs]
def autocomplete(
self, query: str, limit: int = 10, country: str | None = None
) -> list[dict[str, Any]]:
"""Catalog autocomplete — the search-box suggest for a
real-time search UI.
The keystroke endpoint: call it as the user types (it
answers with a small result budget, across artists, albums,
tracks, and playlists). Rows are the *uniform search row*
(``kind``/``id``/``title`` plus ``artist`` and the catalog
garnish when present) — the same row shape :meth:`search`
returns, so a suggest box and a results list render with the
same row code, and the rows' ``kind``/``id`` build links.
Args:
query: The text being typed.
limit: 1..50 suggestions (default 10).
country: Suggest from this market's catalog (ISO country
code) instead of the account's — as in
:meth:`search`. (Without a known country the catalog
answers empty, not an error.)
Returns:
The suggestion rows (``[]`` for a query with no matches).
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: transport failures.
"""
self._ensure_logged_in()
return self._api.autocomplete(query, limit=limit, country=country)
[docs]
def search_all(
self,
query: str,
item_type: str | None = None,
page_size: int = 50,
max_items: int = 200,
country: str | None = None,
) -> dict[str, Any]:
"""Page through the catalog search until it is exhausted.
``search()`` returns one page per group; this walks the pages
(the catalog pages each group by the same ``offset``/``limit``)
and merges the rows, so an indexing client can collect *all*
matches of a query without paging the host. It stops when
every group is exhausted (the per-group totals), a page comes
back empty, a page is short, or ``max_items`` rows have been
collected — the defaults mirror the reference fork's
``searchAll`` (50 per page, 200 rows total).
Result: the same shape as :meth:`search`
(``{"text", "type", "total", "items"}``), with ``items`` the
merged rows (page order, groups interleaved) and ``total`` the
catalog's overall sum (which may still exceed ``len(items)``
when ``max_items`` capped the run).
Args:
query: Search text.
item_type: As in :meth:`search` (``None`` = all groups).
page_size: Rows per request, 1..50 (the catalog's cap).
max_items: Hard cap on merged rows.
country: As in :meth:`search`.
"""
items: list[dict] = []
total = 0
page: dict[str, Any] | None = None
offset = 0
while len(items) < max(1, max_items):
page_limit = min(max(1, page_size), max(1, max_items) - len(items))
page = self.search(
query,
item_type=item_type,
limit=page_limit,
offset=offset,
country=country,
)
rows = page["items"]
if not rows:
break
total = page["total"]
items.extend(rows)
if len(items) >= total or len(rows) < page_limit:
break
offset += page_limit
assert page is not None # the loop always fetches at least one page
wrapper: dict[str, Any] = {
"text": query,
"type": page["type"],
"total": total,
"items": items,
}
# The best-match banner comes from the first page (offset 0) —
# the catalog's topHit for the query, present only when the
# catalog names one (garnish convention).
if "top_hit" in page:
wrapper["top_hit"] = page["top_hit"]
return wrapper
[docs]
def search_albums(
self,
query: str,
include_ep: bool | None = None,
) -> list[dict[str, Any]]:
"""Album search that understands *artist-name* queries.
The catalog ranks album search by title, so searching for an
artist's name ("Radiohead") returns only the albums whose
*titles* match — the artist's actual discography is mostly
hidden. This method fixes that: when the query (case and
punctuation-insensitively) is the name of a catalog artist,
that artist's full discography is fetched and listed first,
followed by the title-matching albums, deduplicated by id.
When no artist matches the name, it falls back to the paged
album search (:meth:`search_all`, album group).
Rows are the uniform :meth:`search` album shape (``kind``,
``id``, ``title``, ``artist``, plus ``version``/``quality``/
``artwork`` when known), so a host can render the result
exactly like an album-search page.
Args:
query: Search text (an artist name, an album title, or
anything else — the fallback handles it).
include_ep: Include EPs/singles in the discography.
Defaults to ``config.include_ep``.
"""
# One artist-search page is enough: an artist-name match has
# to be *in* the top hits for its own name (reference-fork
# SEARCH_PAGE_SIZE semantics).
artist_page = self.search(query, item_type="artist", limit=50)
artist_rows = artist_page["items"]
needle = _normalize_artist_name(query)
matched: list[dict[str, Any]] = []
if len(needle) >= 2:
for row in artist_rows:
if _normalize_artist_name(row.get("title")) == needle:
matched.append(row)
rows: list[dict[str, Any]] = []
seen: set[str] = set()
def add_row(row: dict[str, Any]) -> None:
key = str(row.get("id") or "")
if key and key in seen:
return
if key:
seen.add(key)
rows.append(row)
if matched:
ep = include_ep if include_ep is not None else self.config.include_ep
for row in matched:
artist_id = row.get("id")
if artist_id is None or str(artist_id).strip() == "":
continue
try:
albums = self._api.get_artist_albums(artist_id, include_ep=ep)
except TidalError as exc:
# A dead discography endpoint must not hide the
# title hits that still work (per-artist failure
# isolation; the reference fork logs and continues).
logger.info(
"could not load albums for artist %s: %s", artist_id, exc
)
continue
for album in albums:
if album.id is None:
continue
add_row(_album_search_row(album))
if not matched:
# Name is not a catalog artist: fall back to the paged
# album search (the reference fork uses searchAll here).
fallback = self.search_all(query, item_type="album")
for row in fallback["items"]:
add_row(row)
else:
# Discography first, then the albums whose *titles* match
# the query (one page: anything beyond the top 50 title
# hits is reachable via search_all by the host itself).
title_hits = self.search(query, item_type="album", limit=50)
for row in title_hits["items"]:
add_row(row)
return rows
[docs]
def genres(self) -> list[dict[str, Any]]:
"""The catalog's browse genres (the web app's "Discover" grid).
Each row is ``{"name", "path", "has_playlists", "has_artists",
"has_albums", "has_tracks", "has_videos"}`` plus ``artwork``
(the card thumbnail id — pass it to :meth:`artwork_url`) when
the catalog names one. The ``has_*`` flags tell a browse UI
which tabs a genre card offers: build the card, and for every
offered type call :meth:`genre_items` with the same plural
name.
Returns:
One row per genre (the catalog's fixed list, ~20 genres).
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: catalog failures.
"""
self._ensure_logged_in()
return self._api.get_genres()
[docs]
def genre_items(
self,
genre: str,
item_type: str = "tracks",
limit: int = 25,
offset: int = 0,
) -> dict[str, Any]:
"""Browse one genre's catalog items (the "Discover" tab body).
``genre`` is the genre's ``path`` from :meth:`genres` (its
``name`` works too — they are equal on every catalog genre);
``item_type`` one of ``"tracks"``, ``"albums"``, ``"playlists"``,
``"videos"`` — the same plural names as the genre row's
``has_<item_type>`` flags, so a browse UI iterates ``for
item_type in ("tracks", "albums", "playlists", "videos"): if
genre["has_" + item_type]: ...``.
Result: ``{"genre", "type", "total", "items"}`` where each
item is the *uniform search row* (``kind``/``id``/``title``
plus ``artist`` and the catalog garnish when present) — a
browse tab renders with the same row code as a search
result, and the rows build links straight away.
Args:
genre: The genre path (from :meth:`genres`).
item_type: ``"tracks"`` / ``"albums"`` / ``"playlists"``
/ ``"videos"``.
limit: 1..50 items per page.
offset: Result offset for paging.
Returns:
``{"genre", "type", "total", "items"}``.
Raises:
AuthError: not logged in (after a refresh attempt).
ConfigError: unknown ``item_type``.
NotFoundError: unknown genre.
"""
self._ensure_logged_in()
items, total = self._api.get_genre_items(
genre, item_type=item_type, limit=limit, offset=offset
)
return {
"genre": genre,
"type": item_type,
"total": total,
"items": items,
}
[docs]
def user_playlists(
self,
user_id: int | str | None = None,
limit: int = 25,
offset: int = 0,
) -> dict[str, Any]:
"""A user's playlist links (the "My Playlists" tab).
``user_id`` defaults to the logged-in session's user (the
account's own playlists); pass another user id for that
user's *public* playlists. Rows are the uniform playlist
search rows (``kind``/``id``/``title`` — the id is the
linkable ``uuid``), so a library UI renders them with the
same row code as search/browse and builds
``https://tidal.com/playlist/<uuid>`` links straight away.
Args:
user_id: The user's id; ``None`` (default) is the
logged-in session's user.
limit: 1..50 playlists per page.
offset: Result offset for paging.
Returns:
``{"total", "items"}``.
Raises:
AuthError: not logged in (after a refresh attempt) —
and when no ``user_id`` is given (the "self"
default needs the session).
ConfigError: no ``user_id`` given and no session user.
TidalError: catalog failures.
"""
self._ensure_logged_in()
items, total = self._api.get_user_playlists(user_id, limit=limit, offset=offset)
return {"total": total, "items": items}
[docs]
def list_items(
self,
url: str,
deep: bool = False,
quality: Quality | str | None = None,
) -> list[ItemReport]:
"""Item report for a collection (or single track).
Shallow: catalog facts only (every track item is ``ok``). Deep:
each track's stream is actually resolved, so broken items show
up with a ``failed`` status and the reason — the library
equivalent of the legacy ``--diagnose --json`` output. Deep
items additionally carry the delivered ``codec`` and, when the
path can be computed, the *expected* ``path`` plus ``size``
(bytes already on disk at that path, ``0`` when absent) — the
host gets a full "what would this download do" report.
Args:
url: Any Tidal link (or bare id).
deep: resolve each track's stream (slower, one API call
per item) so broken items surface with a reason.
quality: Quality profile or label to resolve at in deep
mode (default: ``config.quality``).
"""
resource = self._api.resolve(url)
tracks = list(self._api.get_items(resource))
quality = parse_quality(quality) if quality is not None else self.config.quality
reports: list[ItemReport] = []
for index, track in enumerate(tracks, start=1):
if track is None:
# A non-audio slot (music video): counted, never
# downloaded — the catalog's item count must match the
# report's.
reports.append(
ItemReport(
index,
"video",
None,
None,
"skipped",
"video item (audio-only library)",
)
)
continue
if track.id is None:
reports.append(
ItemReport(index, "item", None, None, "skipped", "not a track")
)
continue
if not deep:
reports.append(ItemReport(index, "track", track.id, track.title, "ok"))
continue
try:
stream = self._api.resolve_stream(track.id, quality, track)
except TidalError as exc:
reports.append(
ItemReport(
index,
"track",
track.id,
track.title,
"failed",
reason=str(exc),
code=exc.code,
retryable=exc.retryable,
)
)
continue
report = ItemReport(
index,
"track",
track.id,
track.title,
"ok",
delivered_quality=stream.delivered_quality,
codec=stream.codec or None,
)
# Local file status (the library's --diagnose parity): the
# expected path plus what is already on disk at it.
try:
path = self._pipeline.dest_for(resource, track, stream, index)
except TidalError:
path = None
if path is not None:
report.path = str(path)
if path.exists() and path.stat().st_size > 0:
report.size = path.stat().st_size
reports.append(report)
return reports
# -- download ----------------------------------------------------------------
[docs]
def download(
self,
url: str,
dest: str | Path | None = None,
quality: Quality | str | None = None,
skip_broken: bool | None = None,
on_progress: Callable[[dict[str, Any]], None] | None = None,
cancel_event: threading.Event | None = None,
) -> DownloadReport:
"""Download a track, album, playlist, mix, or artist.
Args:
url: Any Tidal link (or bare id).
dest: Destination directory override (default:
``config.output_dir``).
quality: Quality profile or label (``"master"``, ``"flac"``);
default: ``config.quality``.
skip_broken: ``True`` (default) skips broken items and keeps
going; ``False`` stops the run after the first failure.
Both ways the report is returned — inspect
``report.failures``.
on_progress: Callback receiving plain JSON-safe dicts:
``stage`` (``start``/``item``/``bytes``/``complete``),
``kind``, ``index``, ``total``, ``id``, ``title``,
``status``, ``reason`` plus extras (``bytes``,
``total_bytes``, ``path``). Callback exceptions are
logged, never propagated.
cancel_event: Optional host-owned event; setting it cancels.
(``client.cancel()`` cancels all active operations.)
Returns:
A :class:`~tidalpy.models.DownloadReport` — one
:class:`~tidalpy.models.ItemReport` per track with status
``ok``/``failed``/``skipped``, machine ``code``, and
``retryable``.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: collection-level failures (bad URL, region
lock, rate limit, cancellation).
"""
self._ensure_logged_in()
profile = parse_quality(quality) if quality is not None else None
resource = self._api.resolve(url)
stop_event, owned = self._begin_operation(cancel_event)
try:
return self._pipeline.download(
resource,
dest=dest,
quality=profile,
skip_broken=skip_broken,
on_progress=on_progress,
stop_event=stop_event,
)
finally:
self._end_operation(stop_event, owned)
[docs]
def download_many(
self,
urls: Sequence[str],
quality: Quality | str | None = None,
skip_broken: bool = True,
on_progress: ProgressCallback | None = None,
cancel_event: threading.Event | None = None,
) -> list[DownloadReport]:
"""Download many links in one call (the indexing-client facade).
Hosts that collect links in a loop (a search-results importer,
a watchlist sync) would otherwise repeat the resolve +
pipeline dance per URL and hand-roll their own failure
bookkeeping. This walks the links with the library's
contracts:
- **Per-URL failure isolation**: a bad link (unresolvable,
region locked, rate-limited out, a collection that fails
before its first item) becomes a single-item ``failed``
report carrying the machine ``code`` and ``retryable``
flag, and the walk continues.
- **One report per URL, in input order**, so the host
correlates results to links by position.
- **Sequential runs**: each run already manages its own
quality downgrade, retries, and the server's rate-limit
budget; parallel collections would only fight over that
budget.
- Progress events are the per-run dicts, in order (the flat
event contract of :meth:`download`), each annotated with
the run's ``url`` and its position in the input sequence
(``url_index``) — a batch UI correlates events to links
without hand-rolling its own bookkeeping.
Cancellation (``cancel_event`` or the client's ``stop()``)
ends the walk: the run in flight returns its (cancelled-
marked) report, and the remaining URLs are not attempted —
the returned list covers only the URLs processed.
Args:
urls: Tidal links (or bare ids), in download order.
quality: As in :meth:`download`.
skip_broken: As in :meth:`download`.
on_progress: As in :meth:`download`.
cancel_event: As in :meth:`download`.
Returns:
One :class:`~tidalpy.models.DownloadReport` per processed
URL (see the cancellation note above).
"""
self._ensure_logged_in()
profile = parse_quality(quality) if quality is not None else None
requested = (profile or self.config.quality).value
stop_event, owned = self._begin_operation(cancel_event)
reports: list[DownloadReport] = []
# A batch UI must know which URL an event belongs to: every
# per-run event is annotated with its url and url_index (the
# position in the input sequence). Present only in
# download_many — single-URL runs keep the flat contract.
if on_progress is not None:
def annotated(index: int, url: str, event: dict) -> None:
event["url"] = url
event["url_index"] = index
on_progress(event)
def emit_for(index: int, url: str) -> ProgressCallback | None:
if on_progress is None:
return None
return lambda event: annotated(index, url, event)
try:
for url_index, url in enumerate(urls):
if stop_event.is_set():
break # cancelled: the run in flight already reported
try:
resource = self._api.resolve(url)
except TidalError as exc:
# A link no catalog answers: still a report, so
# the host's position-to-link mapping holds.
kind, _item_id = parse_tidal_url(url)
reports.append(
_failure_report(
kind or "unknown",
None,
None,
f"resolve failed: {exc}",
exc,
requested,
)
)
continue
try:
reports.append(
self._pipeline.download(
resource,
dest=None,
quality=profile,
skip_broken=skip_broken,
on_progress=emit_for(url_index, url),
stop_event=stop_event,
)
)
except TidalError as exc:
data = resource.data
reports.append(
_failure_report(
resource.kind,
resource.title,
getattr(data, "id", None),
f"download failed: {exc}",
exc,
requested,
)
)
finally:
self._end_operation(stop_event, owned)
return reports
# -- streaming (host-owned file I/O) --------------------------------------------
[docs]
def stream(
self,
url: str,
quality: Quality | str | None = None,
preview: bool = False,
cancel_event: threading.Event | None = None,
chunk_size: int | None = None,
) -> MediaStream:
"""Resolve a *track* link to a readable plaintext byte stream.
No disk I/O happens in the library: the host pulls chunks and
writes them wherever it wants (S3 multipart, DB blob, pipe to
ffmpeg). For collections, use :meth:`iter_streams` (one
stream per audio track, in order) or resolve a track link
first (``list_items`` / ``metadata``).
Args:
url: A Tidal track link (or bare track id).
quality: Quality profile or label; default: ``config.quality``.
preview: Accept the ~30-second preview when the catalog
cannot deliver the full track (free-tier accounts,
pre-release/exclusive tracks — even on subscribed
accounts). ``True`` never downgrades a track that
*has* a full stream; it only turns the
``PreviewOnlyError`` into a playable stream with
``info.preview is True`` (the search-UI "play
something" path). Downloads never resolve previews:
a 30-second file is never a valid download result.
cancel_event: Optional host-owned cancellation event.
chunk_size: ``MediaStream`` chunk size (>= 16 KiB);
default: ``DEFAULT_CHUNK_SIZE`` (256 KiB).
Returns:
A :class:`~tidalpy.streaming.MediaStream` (context manager,
iterator of bytes, ``.size``, ``.info``, ``.track``).
Raises:
AuthError: not logged in.
TidalError: resolution failures (bad URL, preview-only
when ``preview`` is false, quality unavailable, ...).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
track = resource.data
if resource.kind != "track" or not isinstance(track, Track):
raise TidalError(
"stream() requires a single track; "
f"got a {resource.kind} link — use list_items() to pick a track",
)
if track.id is None:
raise TidalError("resolved track has no id")
profile = parse_quality(quality) if quality is not None else None
stream_info = self._api.resolve_stream(
track.id, profile or self.config.quality, track, allow_preview=preview
)
stop_event, owned = self._begin_operation(cancel_event)
return MediaStream(
track,
stream_info,
self._transport,
stop_event=stop_event,
chunk_size=chunk_size or DEFAULT_CHUNK_SIZE,
# The stream outlives this call: release the client-level
# registration when the host closes it.
on_close=lambda: self._end_operation(stop_event, owned),
)
[docs]
def playback(
self,
url: str,
quality: Quality | str | None = None,
preview: bool = False,
) -> dict[str, Any]:
"""What will actually be played (the player facade).
Resolves the track's stream without downloading a byte: the
delivered quality, codec, rate/depth, preview flag, and the
expiring CDN link (``url``) for hosts that fetch the bytes
themselves. A player UI shows the quality badge ("192 kHz /
24-bit FLAC") before pressing play; a sync host pre-checks
playability. The resolution is cached per
(account, track, quality, preview) — a subsequent
:meth:`stream` of the same track at the same quality spends
no second rate-limited probe.
Args:
url: A Tidal track link (or bare track id).
quality: Quality profile or label; default:
``config.quality`` (as in :meth:`stream`).
preview: As in :meth:`stream` (accept the ~30-second
preview when the full track is unplayable;
``True`` never downgrades a track that has a full
stream).
Returns:
The resolved stream as a plain JSON-safe dict
(``StreamInfo.to_dict``): ``trackId``, ``url``,
``segments``, ``codec``, ``deliveredQuality``,
``requestedQuality``, ``bandwidth``, ``sampleRate``,
``bitDepth``, ``channels``, ``preview``, ...
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: resolution failures (bad URL, not a track,
preview-only when ``preview`` is false, ...).
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
track = resource.data
if resource.kind != "track" or not isinstance(track, Track):
raise TidalError(
"playback() requires a single track; "
f"got a {resource.kind} link — use list_items() to pick a track",
)
if track.id is None:
raise TidalError("resolved track has no id")
profile = parse_quality(quality) if quality is not None else None
stream_info = self._api.resolve_stream(
track.id, profile or self.config.quality, track, allow_preview=preview
)
return stream_info.to_dict()
[docs]
def iter_streams(
self,
url: str,
quality: Quality | str | None = None,
chunk_size: int | None = None,
cancel_event: threading.Event | None = None,
skip_broken: bool = True,
preview: bool = False,
) -> Iterator[MediaStream]:
"""Readable streams for a link's tracks, in catalog order.
The player-client facade: the :meth:`stream` equivalent for
collections (an album, playlist, mix, artist discography) —
a host that plays a collection in order (a UI, a gapless
player, a live broadcast) without saving anything. Each
yielded :class:`~tidalpy.streaming.MediaStream` is
independent: pull its chunks (or open it as a context
manager) and close it before relying on the next one.
- Tracks come back in the collection's order; non-audio
slots (music videos) are skipped — an audio-only library.
- **Per-track failure isolation**: an unresolvable track
(broken item, region lock, quality unavailable) is logged
and skipped when ``skip_broken`` is true; with it false the
error propagates to the host.
- ``preview=True`` (the "play what you can" path): tracks the
catalog can only serve as a ~30-second preview (free-tier
accounts, pre-release/exclusive tracks) yield a preview
stream with ``info.preview is True`` instead of counting
as broken — the player plays *something* in every slot and
can show a badge. Full streams are still preferred; the
flag never downgrades a playable track.
- Streams are resolved one at a time, lazily, as the host
asks for them — the server's rate-limit budget is shared
with every other operation on this client.
- Cancellation (``cancel_event`` or the client's ``stop()``)
ends the walk: the generator returns after the in-flight
item. An abandoned generator releases the client's
operation registration on close.
Args:
url: A Tidal link (or bare id): a single track yields one
stream; a collection yields one per audio track.
quality: As in :meth:`stream`.
chunk_size: As in :meth:`stream`.
cancel_event: As in :meth:`stream`.
skip_broken: Skip unresolvable tracks instead of raising.
preview: As in :meth:`stream` — accept the ~30-second
preview for tracks that have no full stream.
Yields:
One :class:`~tidalpy.streaming.MediaStream` per playable
track.
Raises:
AuthError: not logged in (after a refresh attempt).
TidalError: resolution failures (bad URL, unknown
resource), and unresolvable tracks when
``skip_broken`` is false.
"""
self._ensure_logged_in()
resource = self._api.resolve(url)
profile = parse_quality(quality) if quality is not None else self.config.quality
if resource.kind == "track" and isinstance(resource.data, Track):
tracks: list[Track | None] = [resource.data]
else:
tracks = list(self._api.get_items(resource))
stop_event, owned = self._begin_operation(cancel_event)
try:
for track in tracks:
if stop_event.is_set():
return
if track is None or track.id is None:
continue # non-audio slot (music video)
try:
stream_info = self._api.resolve_stream(
track.id, profile, track, allow_preview=preview
)
except TidalError:
if not skip_broken:
raise
logger.info(
"iter_streams: skipping unresolvable track %s", track.id
)
continue
yield MediaStream(
track,
stream_info,
self._transport,
stop_event=stop_event,
chunk_size=chunk_size or DEFAULT_CHUNK_SIZE,
)
finally:
self._end_operation(stop_event, owned)
# -- internals -------------------------------------------------------------------
def _ensure_logged_in(self) -> None:
self._oauth.ensure_logged_in()