How can I return progress_hook for yt_dlp using FastAPI to end user?

Question

Relevant portion of my code looks something like this:

@directory_router.get("/youtube-dl/{relative_path:path}", tags=["directory"])
def youtube_dl(relative_path, url, name=""):
    """
    Download
    """

    relative_path, _ = set_path(relative_path)

    logger.info(f"{DATA_PATH}{relative_path}")

    if name:
        name = f"{DATA_PATH}{relative_path}/{name}.%(ext)s"
    else:
        name = f"{DATA_PATH}{relative_path}/%(title)s.%(ext)s"

    ydl_opts = {
        "outtmpl": name,
        # "quiet": True
        "logger": logger,
        "progress_hooks": [yt_dlp_hook],
        # "force-overwrites": True
    }

    with yt.YoutubeDL(ydl_opts) as ydl:
        try:
            ydl.download([url])
        except Exception as exp:
            logger.info(exp)
            return str(exp)

I am using this webhook/end point to allow an angular app to accept url/name input and download file to folder. I am able to logger.info .. etc. output the values of the yt_dlp_hook, something like this:

def yt_dlp_hook(download):
    """
    download Hook

    Args:
        download (_type_): _description_
    """

    global TMP_KEYS

    if download.keys() != TMP_KEYS:
        logger.info(f'Status: {download["status"]}')
        logger.info(f'Dict Keys: {download.keys()}')
        TMP_KEYS = download.keys()
        logger.info(download)

Is there a way to stream a string of relevant variables like ETA, download speed etc. etc. to the front end? Is there a better way to do this?

Answer 1

You could use a Queue object to communicate between the threads. So when you call youtube_dl pass in a Queue that you can add messages inside yt_dlp_hook (you'll need to use partial functions to construct it). You'll be best off using asyncio to run the download at the same time as updating the user something like:

import asyncio
from functools import partial
import threading
from youtube_dl import YoutubeDL
from queue import LifoQueue, Empty


def main():
    # Set the url to download
    url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"

    # Get the current event loop
    loop = asyncio.get_event_loop()

    # Create a Last In First Out Queue to communicate between the threads
    queue = LifoQueue()

    # Create the future which will be marked as done once the file is downloaded
    coros = [youtube_dl(url, queue)]
    future = asyncio.gather(*coros)

    # Start a new thread to run the loop_in_thread function (with the positional arguments passed to it)
    t = threading.Thread(target=loop_in_thread, args=[loop, future])
    t.start()

    # While the future isn't finished yet continue
    while not future.done():
        try:
            # Get the latest status update from the que and print it
            data = queue.get_nowait()
            print(data)
        except Empty as e:
            print("no status updates available")
        finally:
            # Sleep between checking for updates
            asyncio.run(asyncio.sleep(0.1))


def loop_in_thread(loop, future):
    loop.run_until_complete(future)


async def youtube_dl(url, queue, name="temp.mp4"):
    """
    Download
    """

    yt_dlp_hook_partial = partial(yt_dlp_hook, queue)

    ydl_opts = {
        "outtmpl": name,
        "progress_hooks": [yt_dlp_hook_partial],
    }
    with YoutubeDL(ydl_opts) as ydl:
        return ydl.download([url])


def yt_dlp_hook(queue: LifoQueue, download):
    """
    download Hook

    Args:
        download (_type_): _description_
    """
    # Instead of logging the data just add the latest data to the queue
    queue.put(download)


if __name__ == "__main__":
    main()