Advanced Patterns

Production-grade workflows for handling complex scenarios: duplicate detection, error recovery, batch operations, and failure handling.

Important: The VDL library is stateless by default. It does not create databases, logs, or implicit state. Your application owns all persistence, history, and state management. These patterns show how to build robust workflows on top of VDL's stateless execution layer.

Duplicate Detection & Collision Handling

VDL detects when media already exists and applies a collision policy to handle conflicts.

Understanding Duplicate Detection

The model: If a completed media file exists at the target path, the request should be skipped.

VDL recognizes: - Completed media — .mp4, .webm, .mkv, etc. - In-flight files — .part, .ytdl, .tmp, .aria2 (continue/resume) - Sidecar files — .vtt, .srt, .jpg, .json (not media, don't count as "downloaded")

Browser/Finder copies are recognized as duplicates:

Original:  episode.mp4
Matches:   episode (1).mp4  ← Browser copy counter
           episode (2).mp4  ← Finder copy format
           episode copy.mp4 ← Finder copy format

Does NOT match:
           episode 2.mp4    ← Real title, not a copy
           episode-v2.mp4   ← Different version

Collision Policies

When a file exists, use if_exists to choose behavior:

Policy	Behavior	When to Use
SKIP	Return SKIPPED without downloading	Archives, incremental jobs
RENAME	Save as title-2.mp4, title-3.mp4, etc.	Batch collection jobs
OVERWRITE	Replace the existing file	Re-downloads, updates
FAIL	Raise VDLDownloadError	Strict workflows, error tracking

Example: Batch with Smart Deduplication

import asyncio
from pathlib import Path
from vdl import VDLClient, IfExists
from vdl.models import Quality

async def batch_with_dedup():
    client = VDLClient()
    output_dir = Path("~/Downloads/VDL").expanduser()

    urls = [
        "https://example.com/video1",
        "https://example.com/video2",
        "https://example.com/video1",  # Duplicate URL!
    ]

    # Create a plan from URLs
    plan = await client.plan(urls)
    print(f"Planned {len(plan.items)} items")

    # Execute with deduplication
    # - Duplicate URLs will share the same output file
    # - First download completes, second is skipped
    results = await client.execute_plan(
        plan,
        quality=Quality.P720,
        if_exists=IfExists.SKIP,  # Skip if file exists
    )

    completed = [r for r in results if r.status.value == "completed"]
    skipped = [r for r in results if r.status.value == "skipped"]

    print(f"Downloaded: {len(completed)}")
    print(f"Skipped (duplicates): {len(skipped)}")

asyncio.run(batch_with_dedup())

Preflight Duplicate Check

Check before downloading using the duplicate detection service:

from vdl import DownloadRequest, VDLClient
from vdl.models import Quality
from pathlib import Path

async def preflight_check():
    # Library mode: stateless by default
    client = VDLClient()

    request = DownloadRequest(
        id="check-1",
        source_url="https://example.com/video",
        quality=Quality.P720,
        output_dir=Path("~/Downloads/VDL").expanduser(),
    )

    # Check if this would be a duplicate
    decision = await client.execution_service.duplicate_detection.check(request)

    print(f"Action: {decision.action}")          # "skip" or "continue"
    print(f"Reason: {decision.reason}")          # why
    print(f"Target path: {decision.target_path}") # where it would go

    # Now decide whether to download
    if decision.action == "skip":
        print("File already exists, skipping")
    else:
        result = await client.download(request.source_url)

asyncio.run(preflight_check())

---

Error Handling & Recovery

Handle download failures, implement retry logic, and recover gracefully.

Error Types & Classification

VDL provides specific exception types for different failure modes:

from vdl import (
    VDLError,              # Base class
    VDLConfigError,        # Config load/validation
    VDLInputError,         # User input parsing
    VDLDownloadError,      # Download operation
    VDLDependencyError,    # Missing ffmpeg, yt-dlp
    VDLStateError,         # Persistent state issues
    VDLCrawlerError,       # Page crawling
)

async def handle_errors():
    # Library mode: stateless, handle exceptions from your code
    client = VDLClient()

    try:
        result = await client.download("https://example.com/video")
    except VDLDependencyError:
        print("Install ffmpeg and yt-dlp first")
        print("  macOS: brew install ffmpeg yt-dlp")
        print("  Linux: sudo apt-get install ffmpeg && pip install yt-dlp")
    except VDLDownloadError as e:
        print(f"Download failed: {e}")
    except VDLError as e:
        print(f"VDL error: {e}")

Result-Based Error Checking

Most operations return a result with status, not exceptions:

from vdl.models import DownloadStatus

async def check_result():
    client = VDLClient()
    result = await client.download("https://example.com/video")

    if result.status == DownloadStatus.COMPLETED:
        print(f"✓ Downloaded to {result.output_path}")

    elif result.status == DownloadStatus.SKIPPED:
        print(f"⊘ Already present: {result.output_path}")

    elif result.status == DownloadStatus.FAILED:
        print(f"✗ Failed: {result.error}")

        # Check if retryable
        if result.metadata.get("retryable"):
            strategy = result.metadata.get("suggested_retry")
            print(f"Suggestion: {strategy}")

    elif result.status == DownloadStatus.CANCELLED:
        print("Cancelled by user")

Quality Fallback Strategy

Try multiple qualities before failing:

from vdl.models import Quality, DownloadStatus

async def download_with_fallback():
    client = VDLClient()
    url = "https://example.com/video"

    # Fallback chain: prefer high quality, accept lower
    qualities = [Quality.P1080, Quality.P720, Quality.P480, Quality.AUDIO]

    for quality in qualities:
        print(f"Trying {quality.value}...", end=" ")
        result = await client.download(url, quality=quality)

        if result.status == DownloadStatus.COMPLETED:
            print(f"✓ Success!")
            return result
        elif result.status == DownloadStatus.SKIPPED:
            print(f"⊘ Already present")
            return result
        else:
            print(f"✗ Failed")

    raise Exception("All quality levels failed")

asyncio.run(download_with_fallback())

Exponential Backoff Retry

Implement smart retry with increasing delays:

import asyncio
from vdl.models import DownloadStatus

async def retry_with_backoff():
    client = VDLClient()
    url = "https://example.com/video"
    max_retries = 3
    base_delay = 1  # seconds

    for attempt in range(max_retries):
        print(f"Attempt {attempt + 1}/{max_retries}...", end=" ")
        result = await client.download(url)

        if result.status in (DownloadStatus.COMPLETED, DownloadStatus.SKIPPED):
            print("✓ Success!")
            return result

        if not result.metadata.get("retryable"):
            print("✗ Not retryable")
            return result

        if attempt < max_retries - 1:
            delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, ...
            print(f"✗ Failed, retrying in {delay}s...")
            await asyncio.sleep(delay)

    print(f"✗ Failed after {max_retries} attempts")
    return result

asyncio.run(retry_with_backoff())

---

Batch Operations with Failure Isolation

Download multiple items while isolating failures — one failure doesn't stop the batch.

Basic Batch

async def batch_download():
    client = VDLClient()

    urls = [
        "https://example.com/video1",
        "https://example.com/video2",
        "https://example.com/video3",
    ]

    # Plan the batch
    plan = await client.plan(urls)

    # Execute: each item fails independently
    results = await client.execute_plan(plan, quality=Quality.P720)

    # Analyze results
    completed = [r for r in results if r.status.value == "completed"]
    failed = [r for r in results if r.status.value == "failed"]

    print(f"Completed: {len(completed)}/{len(results)}")
    print(f"Failed: {len(failed)}/{len(results)}")

    # Don't stop on first failure — report all
    for result in failed:
        print(f"✗ {result.source_url}: {result.error}")

Batch Summary

Get statistics without iterating:

async def batch_summary():
    client = VDLClient()

    plan = await client.plan([...])

    # execute_batch returns a summary, not individual results
    summary = await client.execute_batch(plan, quality=Quality.P720)

    print(f"✓ Completed: {summary.completed}")
    print(f"⊘ Skipped: {summary.skipped}")
    print(f"✗ Failed: {summary.failed}")
    print(f"⏹ Cancelled: {summary.cancelled}")
    print(f"Time: {summary.elapsed_seconds}s")

Retry Failures from Batch

Use the retry helpers:

from vdl.core.retry import retryable_failed_results, suggested_retry_strategy

async def batch_with_retry():
    client = VDLClient()

    # Initial batch
    plan = await client.plan([...])
    results = await client.execute_plan(plan, quality=Quality.P720, write_subtitles=True)

    # Find retryable failures
    failed = retryable_failed_results(results)
    if not failed:
        print("No retryable failures")
        return results

    # Get the most common suggested retry
    strategy = suggested_retry_strategy(results)
    print(f"Retrying {len(failed)} items: {strategy}")

    # Retry based on suggested strategy
    if strategy == "download without subtitles":
        retry_results = await client.execute_plan(
            plan,
            quality=Quality.P720,
            write_subtitles=False,
        )
        return retry_results

    return results

asyncio.run(batch_with_retry())

---

Adding State to the Stateless Library

The VDL library is stateless by default — it doesn't create databases, logs, or app directories. Your application owns all persistence decisions.

When to add state: - Building a persistent download history - Tracking which URLs have been attempted - Storing user preferences - Maintaining a media library catalog

How to add state: 1. Build your own database/state model 2. Use VDL's stateless API 3. Save results to your state after each operation 4. Query your state before downloading (for deduplication)

Example:

import asyncio
from vdl import VDLClient, DownloadStatus

class YourDownloadService:
    def __init__(self, db_path: str):
        self.client = VDLClient()  # stateless
        self.db = YourDatabase(db_path)  # your persistence layer

    async def download(self, url: str):
        # Check your state before downloading
        if self.db.has_completed(url):
            return "Already downloaded"

        # Use stateless VDL
        result = await self.client.download(url)

        # Save result to your state
        if result.status == DownloadStatus.COMPLETED:
            self.db.record_download(url, result.output_path)

        return result

asyncio.run(YourDownloadService("/path/to/db").download("..."))

---

Using Interactive Mode (With Built-In State)

When you need history/state built-in, use interactive mode instead of the library:

Query Download History

from vdl.runtime import build_interactive_components

async def check_history():
    # Interactive mode wires SQLiteStateStore
    components = build_interactive_components()
    client = VDLClient(state_store=components.state_store)

    # Query all downloads
    all_records = await client.status()
    print(f"Total downloads: {len(all_records)}")

    # Query specific download
    record = await client.status(id="some-download-id")
    if record:
        print(f"Status: {record.status}")
        print(f"Downloaded: {record.completed_at}")
        print(f"Path: {record.output_path}")

asyncio.run(check_history())

Reconciliation: Detect Stale State

When disk and DB are out of sync:

async def reconcile_state():
    components = build_interactive_components()
    client = VDLClient(state_store=components.state_store)

    # Compare DB state with filesystem
    report = await client.reconciliation()

    print(f"Missing files: {len(report.missing)}")
    for entry in report.missing:
        print(f"  ✗ {entry.output_path} (DB says downloaded, but missing on disk)")

    print(f"Orphaned files: {len(report.orphaned)}")
    for entry in report.orphaned:
        print(f"  ? {entry} (on disk, but not in DB)")

asyncio.run(reconcile_state())

---

Common Production Patterns

Pattern 1: Incremental Archive

Download new items into an archive, skip previously downloaded:

async def incremental_archive():
    client = VDLClient()
    archive_dir = Path("/media/archive")

    # New URLs to check
    urls = [...]

    # Execute with SKIP: only download new items
    plan = await client.plan(urls)
    results = await client.execute_plan(
        plan,
        output_dir=archive_dir,
        if_exists=IfExists.SKIP,  # Don't re-download
    )

    completed = len([r for r in results if r.status.value == "completed"])
    skipped = len([r for r in results if r.status.value == "skipped"])

    print(f"Added {completed} new items")
    print(f"Already have {skipped} items")

Pattern 2: Strict Batch (Fail on Collision)

For pipelines where duplicates are unexpected:

async def strict_batch():
    client = VDLClient()

    plan = await client.plan(urls)

    try:
        results = await client.execute_plan(
            plan,
            if_exists=IfExists.FAIL,  # Error if file exists
        )
        print("✓ All downloads succeeded (no duplicates found)")
    except VDLDownloadError as e:
        print(f"✗ Collision detected: {e}")
        # Handle: maybe archive the batch first, retry later

Pattern 3: Graceful Degradation

Try best quality, fall back to audio if video fails:

async def graceful_degradation():
    client = VDLClient()
    url = "https://example.com/video"

    # Try video at various qualities
    for quality in [Quality.P1080, Quality.P720, Quality.P480]:
        result = await client.download(url, quality=quality)
        if result.status == DownloadStatus.COMPLETED:
            return result

    # Fall back to audio
    result = await client.download(url, quality=Quality.AUDIO)
    if result.status == DownloadStatus.COMPLETED:
        print("✓ Downloaded audio (video unavailable)")
        return result

    # Give up
    raise Exception("Unable to download at any quality")

Pattern 4: Rate-Limited Batch

Process items with delays to avoid rate limiting:

import asyncio

async def rate_limited_batch():
    client = VDLClient()
    urls = [...]
    delay_between = 5  # seconds

    for i, url in enumerate(urls):
        if i > 0:
            print(f"Waiting {delay_between}s before next item...")
            await asyncio.sleep(delay_between)

        print(f"Downloading {i+1}/{len(urls)}...", end=" ")
        result = await client.download(url, quality=Quality.P720)
        print(result.status.value)

---

See Also

• RETRY_AND_DUPLICATES.md — detailed semantics
• EXAMPLES.md — basic usage patterns
• demos/advanced_duplicate_detection.py — runnable demo
• demos/advanced_error_handling.py — runnable demo