Building a YouTube Downloader with FastAPI and yt-dlp

In this comprehensive guide, we'll build a modern YouTube video downloader using FastAPI and yt-dlp. Our application will feature a clean web interface, real-time download progress tracking, and robust error handling to deal with YouTube's evolving anti-scraping measures.

🎯 What We'll Build

• Modern Web Interface: Clean, responsive UI with Bootstrap
• Real-time Progress Tracking: Live download progress updates
• Format Selection: Choose from various video/audio quality options
• Task Management: View, monitor, and manage download tasks
• Error Handling: Robust handling of YouTube's restrictions
• File Management: Secure file serving and downloads

🛠️ Tech Stack

• Backend: FastAPI (Python)
• Video Processing: yt-dlp
• Database: SQLite with SQLAlchemy
• Frontend: HTML, CSS, JavaScript, Bootstrap
• Real-time Updates: Polling-based progress tracking

📋 Prerequisites

Before we start, make sure you have:

# Python 3.8+
python --version

# Required packages
pip install fastapi uvicorn yt-dlp sqlalchemy python-multipart jinja2

🏗️ Project Structure

yt-dlp-web/
├── app/
│   ├── main.py              # FastAPI application
│   ├── models.py            # Database models
│   ├── database.py          # Database configuration
│   └── youtube_downloader.py # Core download logic
├── static/
│   ├── script.js           # Frontend JavaScript
│   └── style.css           # Custom styles
├── templates/
│   └── index.html          # Main HTML template
├── downloads/              # Downloaded files directory
└── requirements.txt        # Python dependencies

🗄️ Database Models

Let's start by defining our database models:

# app/models.py
from sqlalchemy import Column, Integer, String, Float, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
from enum import Enum

Base = declarative_base()

class TaskStatus(str, Enum):
    PENDING = "pending"
    DOWNLOADING = "downloading"
    COMPLETED = "completed"
    FAILED = "failed"

class DownloadTask(Base):
    __tablename__ = "download_tasks"
    
    id = Column(Integer, primary_key=True, index=True)
    url = Column(String, nullable=False)
    title = Column(String, nullable=True)
    status = Column(String, default=TaskStatus.PENDING)
    progress = Column(Float, default=0.0)
    file_path = Column(String, nullable=True)
    file_size = Column(Integer, nullable=True)
    error_message = Column(Text, nullable=True)
    format_id = Column(String, nullable=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
    
    def to_dict(self):
        return {
            "id": self.id,
            "url": self.url,
            "title": self.title,
            "status": self.status,
            "progress": self.progress,
            "file_path": self.file_path,
            "file_size": self.file_size,
            "error_message": self.error_message,
            "format_id": self.format_id,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None
        }

🎬 YouTube Downloader Core

The heart of our application is the YouTube downloader class:

# app/youtube_downloader.py
import yt_dlp
import os
import threading
from typing import Dict, Any, Callable, Optional
from models import TaskStatus

class YouTubeDownloader:
    def __init__(self, download_dir: str = "./downloads"):
        self.download_dir = download_dir
        
    def get_video_info(self, url: str) -> Dict[str, Any]:
        """Extract video information without downloading"""
        ydl_opts = {
            'quiet': True,
            'no_warnings': True,
            'extract_flat': False,
            'extractor_args': {
                'youtube': {
                    'player_client': ['tv', 'mweb', 'web'],
                }
            },
            'http_headers': {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            },
            'ignoreerrors': True
        }
        
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=False)
            
            # Process and categorize available formats
            formats = self._process_formats(info.get('formats', []))
            
            return {
                'title': info.get('title', 'Unknown'),
                'duration': info.get('duration', 0),
                'uploader': info.get('uploader', 'Unknown'),
                'view_count': info.get('view_count', 0),
                'thumbnail': info.get('thumbnail', ''),
                'webpage_url': info.get('webpage_url', url),
                'formats': formats
            }
    
    def download_video(self, task_id: int, url: str, format_id: str = 'best', 
                      progress_callback: Optional[Callable] = None):
        """Download video in a separate thread"""
        def download_thread():
            try:
                # Ensure download directory exists
                if not os.path.exists(self.download_dir):
                    os.makedirs(self.download_dir)
                
                # Configure yt-dlp options
                ydl_opts = {
                    'format': self._get_format_selector(format_id),
                    'outtmpl': os.path.join(self.download_dir, "%(title)s-%(id)s.%(ext)s"),
                    'writeinfojson': True,
                    'merge_output_format': 'mp4',
                    'prefer_ffmpeg': True,
                    'extractor_args': {
                        'youtube': {
                            'player_client': ['tv', 'mweb', 'web'],
                        }
                    },
                    'retries': 3,
                    'fragment_retries': 3,
                    'socket_timeout': 30,
                }
                
                if progress_callback:
                    ydl_opts['progress_hooks'] = [self._progress_hook(task_id, progress_callback)]
                
                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                    ydl.download([url])
                    
            except Exception as e:
                if progress_callback:
                    progress_callback(task_id, 0, TaskStatus.FAILED, None, str(e))
        
        thread = threading.Thread(target=download_thread)
        thread.daemon = True
        thread.start()
        return thread
    
    def _progress_hook(self, task_id: int, progress_callback: Callable):
        """Progress tracking hook for yt-dlp"""
        def hook(d):
            if d['status'] == 'downloading':
                if 'total_bytes' in d:
                    progress = (d['downloaded_bytes'] / d['total_bytes']) * 100
                elif 'total_bytes_estimate' in d:
                    progress = (d['downloaded_bytes'] / d['total_bytes_estimate']) * 100
                else:
                    progress = 0
                
                progress_callback(task_id, progress, TaskStatus.DOWNLOADING)
                
            elif d['status'] == 'finished':
                filename = d.get('filename')
                if filename:
                    progress_callback(task_id, 100.0, TaskStatus.COMPLETED, filename)
                
        return hook

🚀 FastAPI Application

Now let's build our FastAPI application:

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from typing import List, Optional
import os

from database import get_db, create_tables, DOWNLOAD_DIR
from models import DownloadTask, TaskStatus
from youtube_downloader import YouTubeDownloader
from pydantic import BaseModel

# Initialize FastAPI app
app = FastAPI(title="YouTube Downloader", description="Modern YouTube video downloader")

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize components
downloader = YouTubeDownloader(DOWNLOAD_DIR)
templates = Jinja2Templates(directory="templates")

# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")

# Pydantic models for API
class VideoInfoRequest(BaseModel):
    url: str

class DownloadRequest(BaseModel):
    url: str
    format_id: Optional[str] = "best"

# Progress callback function
def update_task_progress(task_id: int, progress: float, status: TaskStatus, 
                        file_path: Optional[str] = None, error_message: Optional[str] = None):
    """Update task progress in database"""
    db = next(get_db())
    try:
        task = db.query(DownloadTask).filter(DownloadTask.id == task_id).first()
        if task:
            task.progress = progress
            task.status = status
            if file_path:
                # Handle file path properly
                if os.path.isabs(file_path):
                    task.file_path = os.path.relpath(file_path, DOWNLOAD_DIR)
                else:
                    task.file_path = file_path
                
                # Get file size
                full_path = os.path.join(DOWNLOAD_DIR, task.file_path)
                if os.path.exists(full_path):
                    task.file_size = os.path.getsize(full_path)
                    
            if error_message:
                task.error_message = error_message
            db.commit()
    finally:
        db.close()

# API Routes
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    """Serve the main page"""
    return templates.TemplateResponse("index.html", {"request": request})

@app.post("/api/video-info")
async def get_video_info(request: VideoInfoRequest):
    """Get video information"""
    try:
        info = downloader.get_video_info(request.url)
        return {"success": True, "data": info}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/api/download")
async def start_download(request: DownloadRequest, db: Session = Depends(get_db)):
    """Start download task"""
    try:
        # Get video info first
        video_info = downloader.get_video_info(request.url)
        
        # Create task record
        task = DownloadTask(
            url=request.url,
            title=video_info.get('title'),
            format_id=request.format_id,
            status=TaskStatus.PENDING
        )
        db.add(task)
        db.commit()
        db.refresh(task)
        
        # Start download
        downloader.download_video(
            task.id, 
            request.url, 
            request.format_id,
            update_task_progress
        )
        
        return {"success": True, "task_id": task.id, "message": "Download started"}
        
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/api/tasks")
async def get_tasks(db: Session = Depends(get_db)) -> List[dict]:
    """Get all download tasks"""
    tasks = db.query(DownloadTask).order_by(DownloadTask.created_at.desc()).all()
    return [task.to_dict() for task in tasks]

@app.get("/api/download-file/{task_id}")
async def download_file(task_id: int, db: Session = Depends(get_db)):
    """Serve downloaded file"""
    task = db.query(DownloadTask).filter(DownloadTask.id == task_id).first()
    if not task or not task.file_path:
        raise HTTPException(status_code=404, detail="File not found")
    
    file_path = os.path.join(DOWNLOAD_DIR, task.file_path)
    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="File not found")
    
    return FileResponse(
        file_path,
        filename=os.path.basename(file_path),
        media_type='application/octet-stream'
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

4. Download Web Preview

When running the script, you should see output like:

Analyze video url

Select your preferred format

Check download progress

5. Performance Testing

Monitor server performance during downloads:

# Check server logs for performance metrics
INFO:     "POST /api/video-info HTTP/1.1" 200 OK
INFO:     "POST /api/download HTTP/1.1" 200 OK
INFO:     "GET /api/tasks/1 HTTP/1.1" 200 OK
📥 Starting download for task 1
🔗 URL: https://www.youtube.com/watch?v=dQw4w9WgXcQ
📋 Format: worst
📂 Output template: ./downloads/%(title)s-%(id)s.%(ext)s
🚀 Starting yt-dlp download...
✅ File saved: ./downloads/Rick Astley - Never Gonna Give You Up-dQw4w9WgXcQ.mp4 (Size: 3847291 bytes)
✅ Download completed for task 1

🔮 Future Enhancements

• Playlist Support: Download entire playlists
• User Authentication: Multi-user support
• Download Scheduling: Queue and schedule downloads
• Cloud Storage: Integration with cloud storage services
• Mobile App: React Native or Flutter mobile app
• Docker Support: Containerized deployment

📝 Conclusion

We've built a comprehensive YouTube downloader that combines the power of FastAPI and yt-dlp with a modern web interface. The application handles YouTube's evolving restrictions gracefully and provides a smooth user experience with real-time progress tracking.

The modular architecture makes it easy to extend with additional features like playlist support, user authentication, or cloud storage integration. The robust error handling ensures the application remains stable even when facing YouTube's anti-scraping measures.

🔗 Resources

• This Project Code
• FastAPI Documentation
• yt-dlp GitHub Repository
• SQLAlchemy Documentation
• Bootstrap Documentation

---

Happy coding! 🚀, see you🚀