Django / Celery integration

tidalpy is built for the tidal_transfer migration: it replaces the old CLI-subprocess layer with a structured, in-process API. This page shows the idiomatic patterns.

One client per task

from celery import Celery
from tidalpy import Config, TidalClient
from tidalpy.errors import AuthError, RateLimitError, TidalError

app = Celery("tidal_transfer")

def make_client() -> TidalClient:
    # Token lives in the DB via a custom TokenStore, or on a
    # shared volume via the default FileTokenStore.
    return TidalClient(Config.from_dict(settings.TIDALPY_CONFIG))


@app.task(bind=True, max_retries=3)
def download_collection(task, url, quality="LOSSLESS"):
    client = make_client()
    try:
        if not client.auth_status()["logged_in"]:
            raise AuthError("no valid Tidal login in the DB")
        report = client.download(
            url,
            quality=quality,
            on_progress=lambda e: _progress(task, e),
        )
        return report.to_dict()
    except RateLimitError as exc:
        # Tidal's 429: retry with a backoff sized to the task.
        raise task.retry(exc=exc, countdown=120)
    finally:
        client.close()


def _progress(task, event: dict) -> None:
    # Events are JSON-safe dicts — straight into update_state.
    if event["stage"] == "bytes":
        task.update_state(
            state="PROGRESS",
            meta={
                "current": event["index"],
                "total": event["total"],
                "bytes": event.get("bytes"),
            },
        )
    elif event["stage"] == "complete":
        task.update_state(state="SUCCESS", meta=event)

Cancellation

Map Celery’s on_revoke (or your UI’s “cancel” button) to a shared event:

import threading

_active: dict[str, threading.Event] = {}

def cancel_download(task_id: str) -> None:
    _active.get(task_id, threading.Event()).set()

# inside the task:
stop = threading.Event()
_active[task.request.id] = stop
try:
    report = client.download(url, cancel_event=stop)
finally:
    _active.pop(task.request.id, None)

The task then ends with report.cancelled is True and the partial result — no dangling process, and the next run resumes the .part files.

Host-owned storage (streaming)

When tidal_transfer stores media in S3 (or a DB), skip the disk entirely:

def stream_track(track_url: str, s3_client, bucket: str) -> str:
    client = make_client()
    try:
        with client.stream(track_url) as media:
            return _s3_multipart_upload(
                s3_client, bucket, f"tidal/{media.track_id}.flac",
                media.chunks(),
            )
    finally:
        client.close()

See Streaming (host-owned file I/O) for the full multipart skeleton.

Run status for an admin UI

DownloadReport.summary() is the one-shot status line for a run — counts, bytes, and the failures’ identity, JSON-safe for a task result or an admin column (no item walk on the host):

result = report.summary()
# {"ok": False, "total": 12, "okCount": 10, "skippedCount": 1,
#  "failedCount": 1, "totalBytes": 482133760,
#  "failures": [{"id": 555, "title": "…", "code": "stream",
#                "reason": "…"}], …}

Real-time search endpoint

The search/browse facades back a live search box: the rows are uniform JSON-safe dicts (kind/id/title plus the catalog garnish), so one row serializer covers the suggest box (autocomplete), full results (search), and the genre browse tabs (genres + genre_items):

@app.route("/api/tidal/search")
def tidal_search(q: str):
    client = make_client()
    try:
        if len(q.strip()) < 3:
            return jsonify(client.autocomplete(q))
        page = client.search(q)
        return jsonify(page["items"])
    finally:
        client.close()

A track page’s credits tab is client.track_contributors(url), and the account’s own playlist links are client.user_playlists().

Prefork safety

• Build the client inside the task, never at module import time (a forked worker would inherit the parent’s HTTP session and stop events).

• Two clients in one process are fully independent (no shared credentials, no shared token state).

• The only shared resources are the token store (use the DB-backed store under multi-worker) and, if you keep files, the output directory (safe: promotion is os.replace).