Skip to content

Common VDL Workflows

Working examples for the most common VDL use cases. All examples assume from vdl import * for brevity.

Basic: Download a Single Video

import asyncio
from vdl import VDLClient
from vdl.models import Quality

async def main():
    client = VDLClient()
    result = await client.download(
        "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
        quality=Quality.P720,
    )
    print(result)

asyncio.run(main())

Result fields: - status — DownloadStatus.COMPLETED, SKIPPED, FAILED, or CANCELLED - output_path — Local file path if successful - error — Error message if failed - metadata — Extra info (title, duration, size)

Download with Quality Fallback

async def download_with_fallback():
    client = VDLClient()

    # Try 1080p, fallback to 720p, then audio
    for quality in [Quality.P1080, Quality.P720, Quality.AUDIO]:
        result = await client.download(
            "https://example.com/video",
            quality=quality,
        )
        if result.status == DownloadStatus.COMPLETED:
            return result

    raise Exception("All quality levels failed")

Download with Collision Handling

from vdl import IfExists

async def download_and_handle_duplicates():
    client = VDLClient()

    # IfExists.SKIP: skip if file exists
    # IfExists.RENAME: rename to episode-2.mp4, episode-3.mp4, etc.
    # IfExists.OVERWRITE: replace existing file
    # IfExists.FAIL: raise an error

    result = await client.download(
        "https://example.com/video",
        output_dir="/path/to/videos",
        if_exists=IfExists.RENAME,
    )
    return result

Resolve a Playlist (Inspect Only)

async def inspect_playlist():
    client = VDLClient()

    # Fetch playlist metadata without downloading
    playlist = await client.resolve_playlist(
        "https://www.youtube.com/playlist?list=PLXYZ"
    )

    # Inspect items
    print(f"Title: {playlist.title}")
    print(f"Items: {len(playlist.items)}")
    for item in playlist.items:
        print(f"  - {item.title} ({item.url})")

    return playlist

Download Entire Playlist

async def download_playlist():
    client = VDLClient()

    results = await client.download_playlist(
        "https://www.youtube.com/playlist?list=PLXYZ",
        quality=Quality.P720,
        on_progress=lambda p: print(f"{p.title}: {p.percent}%"),
    )

    # Iterate results
    for result in results:
        if result.status == DownloadStatus.COMPLETED:
            print(f"✓ {result.output_path}")
        elif result.status == DownloadStatus.SKIPPED:
            print(f"⊘ Already present: {result.output_path}")
        else:
            print(f"✗ Failed: {result.error}")

Plan an Input Locally (No Network)

async def plan_input():
    client = VDLClient()

    # Classify input without making network calls
    plan = await client.plan("https://example.com/video")

    # Plan tells you what type of input it is
    print(f"Kind: {plan.kind}")  # URL, PLAYLIST, FILE, etc.
    print(f"Items: {len(plan.items)}")

    return plan

Discover Media on a Web Page

async def discover_and_download():
    client = VDLClient()

    # Crawl a page for downloadable media
    plan = await client.discover(
        "https://example.com/gallery",
        on_status=lambda s: print(f"Crawling: {s}"),
    )

    print(f"Found {len(plan.items)} items")

    # Download all discovered items
    results = await client.execute_plan(
        plan,
        quality=Quality.P720,
        on_progress=lambda p: print(f"{p.title}: {p.percent}%"),
    )

    return results

Download with Progress Callback

async def download_with_progress():
    client = VDLClient()

    async def on_progress(progress: DownloadProgress):
        if progress.percent:
            print(f"{progress.title}: {progress.percent}%")
        if progress.speed_mbps:
            print(f"  Speed: {progress.speed_mbps:.1f} MB/s")
        if progress.eta_seconds:
            print(f"  ETA: {progress.eta_seconds}s")

    result = await client.download(
        "https://example.com/video",
        on_progress=on_progress,
    )

    return result

Cancellable Download

import asyncio
from vdl.cancellation import CancellationToken

async def cancellable_download():
    client = VDLClient()
    token = CancellationToken()

    # In another coroutine, you can cancel:
    async def cancel_after_10_seconds():
        await asyncio.sleep(10)
        token.cancel()

    # Start cancellation task
    asyncio.create_task(cancel_after_10_seconds())

    # Download with token
    result = await client.download(
        "https://example.com/video",
        cancel_token=token,
    )

    if result.status == DownloadStatus.CANCELLED:
        print("User cancelled the download")

    return result

Plan + Execute Workflow

async def plan_then_execute():
    client = VDLClient()

    # Step 1: Plan the input
    plan = await client.plan("URLs.txt")  # File or URL
    print(f"Planning found {len(plan.items)} items")

    # Step 2: (Optional) Filter or inspect plan items
    filtered_items = [
        item for item in plan.items
        if "trailer" not in item.title.lower()
    ]

    # Step 3: Execute the plan
    results = await client.execute_plan(
        plan,
        quality=Quality.P720,
        if_exists=IfExists.SKIP,
    )

    return results

Download Batch with Summary

async def batch_with_summary():
    client = VDLClient()

    # Download and get summary stats
    summary = await client.execute_batch(
        plan=some_plan,
        quality=Quality.P720,
    )

    print(f"Completed: {summary.completed}")
    print(f"Skipped: {summary.skipped}")
    print(f"Failed: {summary.failed}")
    print(f"Total time: {summary.elapsed_seconds}s")

Error Handling

from vdl import (
    VDLError,
    VDLDownloadError,
    VDLConfigError,
    VDLDependencyError,
)

async def safe_download():
    client = VDLClient()

    try:
        result = await client.download("https://example.com/video")
    except VDLConfigError as e:
        print(f"Config problem: {e}")
    except VDLDependencyError as e:
        print(f"Missing ffmpeg or yt-dlp: {e}")
    except VDLDownloadError as e:
        print(f"Download failed: {e}")
    except VDLError as e:
        print(f"Other VDL error: {e}")

Custom Configuration

from pathlib import Path
from vdl import VDLConfig

async def custom_config_download():
    config = VDLConfig(
        output_dir=Path("~/Videos/VDL").expanduser(),
        max_concurrent_downloads=4,
        write_subtitles=True,
        subtitle_languages="en,fr,es",
        ffmpeg_path="/usr/local/bin/ffmpeg",
    )

    client = VDLClient(config=config)

    result = await client.download("https://example.com/video")
    return result

Health Check

async def check_system():
    client = VDLClient()

    report = await client.doctor()

    print(f"ffmpeg: {report.ffmpeg_available}")
    print(f"yt-dlp: {report.ytdlp_available}")
    print(f"ffprobe: {report.ffprobe_available}")
    print(f"Config: {report.config_accessible}")
    print(f"Logs: {report.event_log_accessible}")

Query Download History (Interactive Mode Only)

async def check_download_status():
    client = VDLClient()  # Requires interactive wiring with SQLiteStateStore

    # Query all downloads
    all_records = await client.status()

    # Query specific download
    record = await client.status(id="some-download-id")

    # Check if URL was already downloaded
    if record:
        print(f"Downloaded at: {record.completed_at}")
    else:
        print("Not found in history")

Retry Failed Downloads

from vdl.core.retry import retryable_failed_results

async def retry_failures():
    client = VDLClient()

    # Initial batch
    plan = await client.plan("input.txt")
    results = await client.execute_plan(plan)

    # Find retryable failures
    failed = retryable_failed_results(results)
    if failed:
        print(f"Retrying {len(failed)} items without subtitles...")
        retry_results = await client.execute_plan(
            plan,
            write_subtitles=False,
        )

Advanced Topics

For production workflows with error handling, retries, and complex scenarios, see:

See Also