$ -weight: 500;">pip -weight: 500;">install google-api-python-client redis fastapi uvicorn prometheus-client python-dotenv -weight: 500;">pip -weight: 500;">install google-api-python-client redis fastapi uvicorn prometheus-client python-dotenv -weight: 500;">pip -weight: 500;">install google-api-python-client redis fastapi uvicorn prometheus-client python-dotenv import os import time import json import logging import random from typing import Optional, Dict, Any from googleapiclient.discovery import build from googleapiclient.errors import HttpError import redis # Configure module-level logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) class YouTubeAPIClient: \"\"\"Production-ready YouTube Data API v3 client with quota tracking and exponential backoff.\"\"\" # YouTube API quota costs per endpoint (source: https://developers.google.com/youtube/v3/determine_quota_cost) QUOTA_COSTS = { "videos.list": 1, # Cost per video ID requested (max 50 per call) "channels.list": 1, "search.list": 100 # Expensive endpoint, avoid in hot paths } def __init__( self, api_key: str, quota_limit: int = 10000, redis_client: Optional[redis.Redis] = None, max_retries: int = 5 ): self.api_key = api_key self.quota_limit = quota_limit self.redis_client = redis_client self.max_retries = max_retries self.used_quota = 0 self.youtube = build("youtube", "v3", developerKey=self.api_key) def get_video_metadata(self, video_id: str) -> Optional[Dict[str, Any]]: \"\"\"Fetch metadata for a single YouTube video, with caching and quota checks.\"\"\" # Check cache first if Redis is available if self.redis_client: cached = self.redis_client.get(f"yt:video:{video_id}") if cached: logger.debug(f"Cache hit for video {video_id}") return json.loads(cached) # Check if we have remaining quota if self.used_quota >= self.quota_limit: logger.error(f"Quota exceeded. Used {self.used_quota}/{self.quota_limit}") raise QuotaExceededError(f"Daily quota limit of {self.quota_limit} reached") # Calculate quota cost for this request endpoint = "videos.list" quota_cost = self.QUOTA_COSTS[endpoint] if self.used_quota + quota_cost > self.quota_limit: logger.warning(f"Request would exceed quota: {self.used_quota} + {quota_cost} > {self.quota_limit}") return None # Execute request with exponential backoff response = self._execute_with_backoff( self.youtube.videos().list( part="snippet,contentDetails,statistics", id=video_id ), endpoint ) if not response or not response.get("items"): logger.warning(f"No metadata found for video {video_id}") return None video_data = response["items"][0] self.used_quota += quota_cost logger.info(f"Used quota: {self.used_quota}/{self.quota_limit}") # Cache for 1 hour (3600 seconds) to avoid repeated requests if self.redis_client: self.redis_client.setex( f"yt:video:{video_id}", 3600, json.dumps(video_data) ) return video_data def _execute_with_backoff(self, request, endpoint: str) -> Optional[Dict[str, Any]]: \"\"\"Execute API request with exponential backoff for retries.\"\"\" for attempt in range(self.max_retries): try: return request.execute() except HttpError as e: # Quota exceeded error code is 403 with reason quotaExceeded if e.resp.-weight: 500;">status == 403 and "quotaExceeded" in str(e.content): logger.error(f"Quota exceeded on attempt {attempt + 1}") raise QuotaExceededError("YouTube API quota exceeded") from e # Retryable errors: 429 (too many requests), 500, 503 if e.resp.-weight: 500;">status in (429, 500, 503): backoff = 2 ** attempt + random.uniform(0, 1) # Jitter to avoid thundering herd logger.warning(f"Retrying {endpoint} after {backoff}s (attempt {attempt + 1})") time.sleep(backoff) else: logger.error(f"Non-retryable error: {e}") raise except Exception as e: logger.error(f"Unexpected error: {e}") raise logger.error(f"Max retries ({self.max_retries}) exceeded for {endpoint}") return None class QuotaExceededError(Exception): \"\"\"Custom exception for YouTube API quota exceeded errors.\"\"\" pass import os import time import json import logging import random from typing import Optional, Dict, Any from googleapiclient.discovery import build from googleapiclient.errors import HttpError import redis # Configure module-level logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) class YouTubeAPIClient: \"\"\"Production-ready YouTube Data API v3 client with quota tracking and exponential backoff.\"\"\" # YouTube API quota costs per endpoint (source: https://developers.google.com/youtube/v3/determine_quota_cost) QUOTA_COSTS = { "videos.list": 1, # Cost per video ID requested (max 50 per call) "channels.list": 1, "search.list": 100 # Expensive endpoint, avoid in hot paths } def __init__( self, api_key: str, quota_limit: int = 10000, redis_client: Optional[redis.Redis] = None, max_retries: int = 5 ): self.api_key = api_key self.quota_limit = quota_limit self.redis_client = redis_client self.max_retries = max_retries self.used_quota = 0 self.youtube = build("youtube", "v3", developerKey=self.api_key) def get_video_metadata(self, video_id: str) -> Optional[Dict[str, Any]]: \"\"\"Fetch metadata for a single YouTube video, with caching and quota checks.\"\"\" # Check cache first if Redis is available if self.redis_client: cached = self.redis_client.get(f"yt:video:{video_id}") if cached: logger.debug(f"Cache hit for video {video_id}") return json.loads(cached) # Check if we have remaining quota if self.used_quota >= self.quota_limit: logger.error(f"Quota exceeded. Used {self.used_quota}/{self.quota_limit}") raise QuotaExceededError(f"Daily quota limit of {self.quota_limit} reached") # Calculate quota cost for this request endpoint = "videos.list" quota_cost = self.QUOTA_COSTS[endpoint] if self.used_quota + quota_cost > self.quota_limit: logger.warning(f"Request would exceed quota: {self.used_quota} + {quota_cost} > {self.quota_limit}") return None # Execute request with exponential backoff response = self._execute_with_backoff( self.youtube.videos().list( part="snippet,contentDetails,statistics", id=video_id ), endpoint ) if not response or not response.get("items"): logger.warning(f"No metadata found for video {video_id}") return None video_data = response["items"][0] self.used_quota += quota_cost logger.info(f"Used quota: {self.used_quota}/{self.quota_limit}") # Cache for 1 hour (3600 seconds) to avoid repeated requests if self.redis_client: self.redis_client.setex( f"yt:video:{video_id}", 3600, json.dumps(video_data) ) return video_data def _execute_with_backoff(self, request, endpoint: str) -> Optional[Dict[str, Any]]: \"\"\"Execute API request with exponential backoff for retries.\"\"\" for attempt in range(self.max_retries): try: return request.execute() except HttpError as e: # Quota exceeded error code is 403 with reason quotaExceeded if e.resp.-weight: 500;">status == 403 and "quotaExceeded" in str(e.content): logger.error(f"Quota exceeded on attempt {attempt + 1}") raise QuotaExceededError("YouTube API quota exceeded") from e # Retryable errors: 429 (too many requests), 500, 503 if e.resp.-weight: 500;">status in (429, 500, 503): backoff = 2 ** attempt + random.uniform(0, 1) # Jitter to avoid thundering herd logger.warning(f"Retrying {endpoint} after {backoff}s (attempt {attempt + 1})") time.sleep(backoff) else: logger.error(f"Non-retryable error: {e}") raise except Exception as e: logger.error(f"Unexpected error: {e}") raise logger.error(f"Max retries ({self.max_retries}) exceeded for {endpoint}") return None class QuotaExceededError(Exception): \"\"\"Custom exception for YouTube API quota exceeded errors.\"\"\" pass import os import time import json import logging import random from typing import Optional, Dict, Any from googleapiclient.discovery import build from googleapiclient.errors import HttpError import redis # Configure module-level logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) class YouTubeAPIClient: \"\"\"Production-ready YouTube Data API v3 client with quota tracking and exponential backoff.\"\"\" # YouTube API quota costs per endpoint (source: https://developers.google.com/youtube/v3/determine_quota_cost) QUOTA_COSTS = { "videos.list": 1, # Cost per video ID requested (max 50 per call) "channels.list": 1, "search.list": 100 # Expensive endpoint, avoid in hot paths } def __init__( self, api_key: str, quota_limit: int = 10000, redis_client: Optional[redis.Redis] = None, max_retries: int = 5 ): self.api_key = api_key self.quota_limit = quota_limit self.redis_client = redis_client self.max_retries = max_retries self.used_quota = 0 self.youtube = build("youtube", "v3", developerKey=self.api_key) def get_video_metadata(self, video_id: str) -> Optional[Dict[str, Any]]: \"\"\"Fetch metadata for a single YouTube video, with caching and quota checks.\"\"\" # Check cache first if Redis is available if self.redis_client: cached = self.redis_client.get(f"yt:video:{video_id}") if cached: logger.debug(f"Cache hit for video {video_id}") return json.loads(cached) # Check if we have remaining quota if self.used_quota >= self.quota_limit: logger.error(f"Quota exceeded. Used {self.used_quota}/{self.quota_limit}") raise QuotaExceededError(f"Daily quota limit of {self.quota_limit} reached") # Calculate quota cost for this request endpoint = "videos.list" quota_cost = self.QUOTA_COSTS[endpoint] if self.used_quota + quota_cost > self.quota_limit: logger.warning(f"Request would exceed quota: {self.used_quota} + {quota_cost} > {self.quota_limit}") return None # Execute request with exponential backoff response = self._execute_with_backoff( self.youtube.videos().list( part="snippet,contentDetails,statistics", id=video_id ), endpoint ) if not response or not response.get("items"): logger.warning(f"No metadata found for video {video_id}") return None video_data = response["items"][0] self.used_quota += quota_cost logger.info(f"Used quota: {self.used_quota}/{self.quota_limit}") # Cache for 1 hour (3600 seconds) to avoid repeated requests if self.redis_client: self.redis_client.setex( f"yt:video:{video_id}", 3600, json.dumps(video_data) ) return video_data def _execute_with_backoff(self, request, endpoint: str) -> Optional[Dict[str, Any]]: \"\"\"Execute API request with exponential backoff for retries.\"\"\" for attempt in range(self.max_retries): try: return request.execute() except HttpError as e: # Quota exceeded error code is 403 with reason quotaExceeded if e.resp.-weight: 500;">status == 403 and "quotaExceeded" in str(e.content): logger.error(f"Quota exceeded on attempt {attempt + 1}") raise QuotaExceededError("YouTube API quota exceeded") from e # Retryable errors: 429 (too many requests), 500, 503 if e.resp.-weight: 500;">status in (429, 500, 503): backoff = 2 ** attempt + random.uniform(0, 1) # Jitter to avoid thundering herd logger.warning(f"Retrying {endpoint} after {backoff}s (attempt {attempt + 1})") time.sleep(backoff) else: logger.error(f"Non-retryable error: {e}") raise except Exception as e: logger.error(f"Unexpected error: {e}") raise logger.error(f"Max retries ({self.max_retries}) exceeded for {endpoint}") return None class QuotaExceededError(Exception): \"\"\"Custom exception for YouTube API quota exceeded errors.\"\"\" pass import csv import sys import argparse import logging import time from typing import List, Dict, Any from client import YouTubeAPIClient, QuotaExceededError import redis # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) def load_video_ids(csv_path: str) -> List[str]: \"\"\"Load video IDs from a CSV file with a 'video_id' column.\"\"\" video_ids = [] try: with open(csv_path, mode="r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: video_id = row.get("video_id") if video_id and len(video_id) == 11: # Valid YouTube video ID length video_ids.append(video_id) else: logger.warning(f"Invalid video ID: {video_id}") logger.info(f"Loaded {len(video_ids)} valid video IDs from {csv_path}") return video_ids except FileNotFoundError: logger.error(f"CSV file not found: {csv_path}") sys.exit(1) except Exception as e: logger.error(f"Error loading CSV: {e}") sys.exit(1) def batch_fetch_metadata( client: YouTubeAPIClient, video_ids: List[str], batch_size: int = 50 ) -> List[Dict[str, Any]]: \"\"\"Fetch metadata for video IDs in batches.\"\"\" results = [] total_batches = (len(video_ids) + batch_size - 1) // batch_size # Ceiling division for i in range(0, len(video_ids), batch_size): batch = video_ids[i:i + batch_size] batch_num = (i // batch_size) + 1 logger.info(f"Processing batch {batch_num}/{total_batches} ({len(batch)} videos)") try: # Batch request: join video IDs with comma response = client._execute_with_backoff( client.youtube.videos().list( part="snippet,contentDetails,statistics", id=",".join(batch) ), "videos.list" ) if response and response.get("items"): for item in response["items"]: results.append(item) # Update quota usage (1 unit per batch, not per video) client.used_quota += 1 logger.info(f"Batch {batch_num} complete. Used quota: {client.used_quota}/{client.quota_limit}") else: logger.warning(f"No results for batch {batch_num}") except QuotaExceededError as e: logger.error(f"Quota exceeded during batch {batch_num}: {e}") break except Exception as e: logger.error(f"Error processing batch {batch_num}: {e}") continue # Rate limit to avoid hitting YouTube's 100 requests/second limit if batch_num % 10 == 0: logger.info("Rate limiting: sleeping 1s after 10 batches") time.sleep(1) return results def main(): parser = argparse.ArgumentParser(description="Batch fetch YouTube video metadata") parser.add_argument("--csv", required=True, help="Path to CSV file with video_id column") parser.add_argument("--api-key", required=True, help="YouTube Data API v3 key") parser.add_argument("--output", default="results.json", help="Output JSON file path") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", default=6379, type=int, help="Redis port") args = parser.parse_args() # Initialize Redis client redis_client = redis.Redis(host=args.redis_host, port=args.redis_port, decode_responses=True) try: redis_client.ping() except Exception as e: logger.warning(f"Redis unavailable: {e}. Proceeding without cache.") redis_client = None # Initialize YouTube client client = YouTubeAPIClient( api_key=args.api_key, redis_client=redis_client, quota_limit=10000 ) # Load video IDs video_ids = load_video_ids(args.csv) if not video_ids: logger.error("No valid video IDs to process") sys.exit(1) # Fetch metadata results = batch_fetch_metadata(client, video_ids) # Export results with open(args.output, mode="w", encoding="utf-8") as f: json.dump(results, f, indent=2) logger.info(f"Exported {len(results)} results to {args.output}") logger.info(f"Total quota used: {client.used_quota}/{client.quota_limit}") if __name__ == "__main__": main() import csv import sys import argparse import logging import time from typing import List, Dict, Any from client import YouTubeAPIClient, QuotaExceededError import redis # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) def load_video_ids(csv_path: str) -> List[str]: \"\"\"Load video IDs from a CSV file with a 'video_id' column.\"\"\" video_ids = [] try: with open(csv_path, mode="r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: video_id = row.get("video_id") if video_id and len(video_id) == 11: # Valid YouTube video ID length video_ids.append(video_id) else: logger.warning(f"Invalid video ID: {video_id}") logger.info(f"Loaded {len(video_ids)} valid video IDs from {csv_path}") return video_ids except FileNotFoundError: logger.error(f"CSV file not found: {csv_path}") sys.exit(1) except Exception as e: logger.error(f"Error loading CSV: {e}") sys.exit(1) def batch_fetch_metadata( client: YouTubeAPIClient, video_ids: List[str], batch_size: int = 50 ) -> List[Dict[str, Any]]: \"\"\"Fetch metadata for video IDs in batches.\"\"\" results = [] total_batches = (len(video_ids) + batch_size - 1) // batch_size # Ceiling division for i in range(0, len(video_ids), batch_size): batch = video_ids[i:i + batch_size] batch_num = (i // batch_size) + 1 logger.info(f"Processing batch {batch_num}/{total_batches} ({len(batch)} videos)") try: # Batch request: join video IDs with comma response = client._execute_with_backoff( client.youtube.videos().list( part="snippet,contentDetails,statistics", id=",".join(batch) ), "videos.list" ) if response and response.get("items"): for item in response["items"]: results.append(item) # Update quota usage (1 unit per batch, not per video) client.used_quota += 1 logger.info(f"Batch {batch_num} complete. Used quota: {client.used_quota}/{client.quota_limit}") else: logger.warning(f"No results for batch {batch_num}") except QuotaExceededError as e: logger.error(f"Quota exceeded during batch {batch_num}: {e}") break except Exception as e: logger.error(f"Error processing batch {batch_num}: {e}") continue # Rate limit to avoid hitting YouTube's 100 requests/second limit if batch_num % 10 == 0: logger.info("Rate limiting: sleeping 1s after 10 batches") time.sleep(1) return results def main(): parser = argparse.ArgumentParser(description="Batch fetch YouTube video metadata") parser.add_argument("--csv", required=True, help="Path to CSV file with video_id column") parser.add_argument("--api-key", required=True, help="YouTube Data API v3 key") parser.add_argument("--output", default="results.json", help="Output JSON file path") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", default=6379, type=int, help="Redis port") args = parser.parse_args() # Initialize Redis client redis_client = redis.Redis(host=args.redis_host, port=args.redis_port, decode_responses=True) try: redis_client.ping() except Exception as e: logger.warning(f"Redis unavailable: {e}. Proceeding without cache.") redis_client = None # Initialize YouTube client client = YouTubeAPIClient( api_key=args.api_key, redis_client=redis_client, quota_limit=10000 ) # Load video IDs video_ids = load_video_ids(args.csv) if not video_ids: logger.error("No valid video IDs to process") sys.exit(1) # Fetch metadata results = batch_fetch_metadata(client, video_ids) # Export results with open(args.output, mode="w", encoding="utf-8") as f: json.dump(results, f, indent=2) logger.info(f"Exported {len(results)} results to {args.output}") logger.info(f"Total quota used: {client.used_quota}/{client.quota_limit}") if __name__ == "__main__": main() import csv import sys import argparse import logging import time from typing import List, Dict, Any from client import YouTubeAPIClient, QuotaExceededError import redis # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) def load_video_ids(csv_path: str) -> List[str]: \"\"\"Load video IDs from a CSV file with a 'video_id' column.\"\"\" video_ids = [] try: with open(csv_path, mode="r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: video_id = row.get("video_id") if video_id and len(video_id) == 11: # Valid YouTube video ID length video_ids.append(video_id) else: logger.warning(f"Invalid video ID: {video_id}") logger.info(f"Loaded {len(video_ids)} valid video IDs from {csv_path}") return video_ids except FileNotFoundError: logger.error(f"CSV file not found: {csv_path}") sys.exit(1) except Exception as e: logger.error(f"Error loading CSV: {e}") sys.exit(1) def batch_fetch_metadata( client: YouTubeAPIClient, video_ids: List[str], batch_size: int = 50 ) -> List[Dict[str, Any]]: \"\"\"Fetch metadata for video IDs in batches.\"\"\" results = [] total_batches = (len(video_ids) + batch_size - 1) // batch_size # Ceiling division for i in range(0, len(video_ids), batch_size): batch = video_ids[i:i + batch_size] batch_num = (i // batch_size) + 1 logger.info(f"Processing batch {batch_num}/{total_batches} ({len(batch)} videos)") try: # Batch request: join video IDs with comma response = client._execute_with_backoff( client.youtube.videos().list( part="snippet,contentDetails,statistics", id=",".join(batch) ), "videos.list" ) if response and response.get("items"): for item in response["items"]: results.append(item) # Update quota usage (1 unit per batch, not per video) client.used_quota += 1 logger.info(f"Batch {batch_num} complete. Used quota: {client.used_quota}/{client.quota_limit}") else: logger.warning(f"No results for batch {batch_num}") except QuotaExceededError as e: logger.error(f"Quota exceeded during batch {batch_num}: {e}") break except Exception as e: logger.error(f"Error processing batch {batch_num}: {e}") continue # Rate limit to avoid hitting YouTube's 100 requests/second limit if batch_num % 10 == 0: logger.info("Rate limiting: sleeping 1s after 10 batches") time.sleep(1) return results def main(): parser = argparse.ArgumentParser(description="Batch fetch YouTube video metadata") parser.add_argument("--csv", required=True, help="Path to CSV file with video_id column") parser.add_argument("--api-key", required=True, help="YouTube Data API v3 key") parser.add_argument("--output", default="results.json", help="Output JSON file path") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", default=6379, type=int, help="Redis port") args = parser.parse_args() # Initialize Redis client redis_client = redis.Redis(host=args.redis_host, port=args.redis_port, decode_responses=True) try: redis_client.ping() except Exception as e: logger.warning(f"Redis unavailable: {e}. Proceeding without cache.") redis_client = None # Initialize YouTube client client = YouTubeAPIClient( api_key=args.api_key, redis_client=redis_client, quota_limit=10000 ) # Load video IDs video_ids = load_video_ids(args.csv) if not video_ids: logger.error("No valid video IDs to process") sys.exit(1) # Fetch metadata results = batch_fetch_metadata(client, video_ids) # Export results with open(args.output, mode="w", encoding="utf-8") as f: json.dump(results, f, indent=2) logger.info(f"Exported {len(results)} results to {args.output}") logger.info(f"Total quota used: {client.used_quota}/{client.quota_limit}") if __name__ == "__main__": main() import os import logging from typing import Optional, Dict, Any from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.responses import JSONResponse from fastapi.security import APIKeyHeader from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import redis from client import YouTubeAPIClient, QuotaExceededError # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI(title="YouTube Metadata API", version="1.0.0") # Rate limiter: 100 requests/minute per IP limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Prometheus metrics REQUEST_COUNT = Counter( "yt_api_requests_total", "Total API requests", ["endpoint", "-weight: 500;">status"] ) LATENCY = Histogram( "yt_api_latency_seconds", "Request latency in seconds", ["endpoint"] ) CACHE_HIT = Counter( "yt_api_cache_hits_total", "Total cache hits" ) QUOTA_REMAINING = Counter( "yt_api_quota_remaining", "Remaining API quota" ) # API key authentication API_KEY_HEADER = APIKeyHeader(name="X-API-Key") def verify_api_key(api_key: str = Depends(API_KEY_HEADER)): if api_key != os.getenv("API_KEY"): raise HTTPException(status_code=401, detail="Invalid API key") return api_key # Initialize YouTube client redis_client = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"), port=6379, decode_responses=True) youtube_client = YouTubeAPIClient( api_key=os.getenv("YOUTUBE_API_KEY"), redis_client=redis_client, quota_limit=int(os.getenv("QUOTA_LIMIT", 10000)) ) @app.get("/health") async def health_check(): \"\"\"Health check endpoint.\"\"\" try: redis_client.ping() return {"-weight: 500;">status": "healthy", "quota_remaining": youtube_client.quota_limit - youtube_client.used_quota} except Exception as e: return JSONResponse(status_code=503, content={"-weight: 500;">status": "unhealthy", "error": str(e)}) @app.get("/video/{video_id}") @limiter.limit("100/minute") async def get_video(request: Request, video_id: str, api_key: str = Depends(verify_api_key)): \"\"\"Fetch metadata for a single video.\"\"\" with LATENCY.labels(endpoint="/video/{video_id}").time(): try: # Check cache hit if redis_client.get(f"yt:video:{video_id}"): CACHE_HIT.inc() metadata = youtube_client.get_video_metadata(video_id) REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="200").inc() QUOTA_REMAINING.inc(youtube_client.quota_limit - youtube_client.used_quota) if not metadata: raise HTTPException(status_code=404, detail="Video not found") return metadata except QuotaExceededError: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="429").inc() raise HTTPException(status_code=429, detail="API quota exceeded") except Exception as e: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="500").inc() logger.error(f"Error fetching video {video_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.get("/metrics") async def metrics(): \"\"\"Prometheus metrics endpoint.\"\"\" return generate_latest(), 200, {"Content-Type": CONTENT_TYPE_LATEST} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) import os import logging from typing import Optional, Dict, Any from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.responses import JSONResponse from fastapi.security import APIKeyHeader from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import redis from client import YouTubeAPIClient, QuotaExceededError # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI(title="YouTube Metadata API", version="1.0.0") # Rate limiter: 100 requests/minute per IP limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Prometheus metrics REQUEST_COUNT = Counter( "yt_api_requests_total", "Total API requests", ["endpoint", "-weight: 500;">status"] ) LATENCY = Histogram( "yt_api_latency_seconds", "Request latency in seconds", ["endpoint"] ) CACHE_HIT = Counter( "yt_api_cache_hits_total", "Total cache hits" ) QUOTA_REMAINING = Counter( "yt_api_quota_remaining", "Remaining API quota" ) # API key authentication API_KEY_HEADER = APIKeyHeader(name="X-API-Key") def verify_api_key(api_key: str = Depends(API_KEY_HEADER)): if api_key != os.getenv("API_KEY"): raise HTTPException(status_code=401, detail="Invalid API key") return api_key # Initialize YouTube client redis_client = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"), port=6379, decode_responses=True) youtube_client = YouTubeAPIClient( api_key=os.getenv("YOUTUBE_API_KEY"), redis_client=redis_client, quota_limit=int(os.getenv("QUOTA_LIMIT", 10000)) ) @app.get("/health") async def health_check(): \"\"\"Health check endpoint.\"\"\" try: redis_client.ping() return {"-weight: 500;">status": "healthy", "quota_remaining": youtube_client.quota_limit - youtube_client.used_quota} except Exception as e: return JSONResponse(status_code=503, content={"-weight: 500;">status": "unhealthy", "error": str(e)}) @app.get("/video/{video_id}") @limiter.limit("100/minute") async def get_video(request: Request, video_id: str, api_key: str = Depends(verify_api_key)): \"\"\"Fetch metadata for a single video.\"\"\" with LATENCY.labels(endpoint="/video/{video_id}").time(): try: # Check cache hit if redis_client.get(f"yt:video:{video_id}"): CACHE_HIT.inc() metadata = youtube_client.get_video_metadata(video_id) REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="200").inc() QUOTA_REMAINING.inc(youtube_client.quota_limit - youtube_client.used_quota) if not metadata: raise HTTPException(status_code=404, detail="Video not found") return metadata except QuotaExceededError: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="429").inc() raise HTTPException(status_code=429, detail="API quota exceeded") except Exception as e: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="500").inc() logger.error(f"Error fetching video {video_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.get("/metrics") async def metrics(): \"\"\"Prometheus metrics endpoint.\"\"\" return generate_latest(), 200, {"Content-Type": CONTENT_TYPE_LATEST} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) import os import logging from typing import Optional, Dict, Any from fastapi import FastAPI, HTTPException, Depends, Request from fastapi.responses import JSONResponse from fastapi.security import APIKeyHeader from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import redis from client import YouTubeAPIClient, QuotaExceededError # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI(title="YouTube Metadata API", version="1.0.0") # Rate limiter: 100 requests/minute per IP limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Prometheus metrics REQUEST_COUNT = Counter( "yt_api_requests_total", "Total API requests", ["endpoint", "-weight: 500;">status"] ) LATENCY = Histogram( "yt_api_latency_seconds", "Request latency in seconds", ["endpoint"] ) CACHE_HIT = Counter( "yt_api_cache_hits_total", "Total cache hits" ) QUOTA_REMAINING = Counter( "yt_api_quota_remaining", "Remaining API quota" ) # API key authentication API_KEY_HEADER = APIKeyHeader(name="X-API-Key") def verify_api_key(api_key: str = Depends(API_KEY_HEADER)): if api_key != os.getenv("API_KEY"): raise HTTPException(status_code=401, detail="Invalid API key") return api_key # Initialize YouTube client redis_client = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"), port=6379, decode_responses=True) youtube_client = YouTubeAPIClient( api_key=os.getenv("YOUTUBE_API_KEY"), redis_client=redis_client, quota_limit=int(os.getenv("QUOTA_LIMIT", 10000)) ) @app.get("/health") async def health_check(): \"\"\"Health check endpoint.\"\"\" try: redis_client.ping() return {"-weight: 500;">status": "healthy", "quota_remaining": youtube_client.quota_limit - youtube_client.used_quota} except Exception as e: return JSONResponse(status_code=503, content={"-weight: 500;">status": "unhealthy", "error": str(e)}) @app.get("/video/{video_id}") @limiter.limit("100/minute") async def get_video(request: Request, video_id: str, api_key: str = Depends(verify_api_key)): \"\"\"Fetch metadata for a single video.\"\"\" with LATENCY.labels(endpoint="/video/{video_id}").time(): try: # Check cache hit if redis_client.get(f"yt:video:{video_id}"): CACHE_HIT.inc() metadata = youtube_client.get_video_metadata(video_id) REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="200").inc() QUOTA_REMAINING.inc(youtube_client.quota_limit - youtube_client.used_quota) if not metadata: raise HTTPException(status_code=404, detail="Video not found") return metadata except QuotaExceededError: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="429").inc() raise HTTPException(status_code=429, detail="API quota exceeded") except Exception as e: REQUEST_COUNT.labels(endpoint="/video/{video_id}", -weight: 500;">status="500").inc() logger.error(f"Error fetching video {video_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.get("/metrics") async def metrics(): \"\"\"Prometheus metrics endpoint.\"\"\" return generate_latest(), 200, {"Content-Type": CONTENT_TYPE_LATEST} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) # BAD: search.list costs 100 quota search_request = youtube.search().list(part="id", channelId="UC_x5XG1OV2P6uZZ5FSM9Ttw", maxResults=50) # GOOD: channels.list + playlistItems.list costs 2 quota total channel_request = youtube.channels().list(part="contentDetails", id="UC_x5XG1OV2P6uZZ5FSM9Ttw") playlist_id = channel_request.execute()["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"] playlist_request = youtube.playlistItems().list(part="contentDetails", playlistId=playlist_id, maxResults=50) # BAD: search.list costs 100 quota search_request = youtube.search().list(part="id", channelId="UC_x5XG1OV2P6uZZ5FSM9Ttw", maxResults=50) # GOOD: channels.list + playlistItems.list costs 2 quota total channel_request = youtube.channels().list(part="contentDetails", id="UC_x5XG1OV2P6uZZ5FSM9Ttw") playlist_id = channel_request.execute()["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"] playlist_request = youtube.playlistItems().list(part="contentDetails", playlistId=playlist_id, maxResults=50) # BAD: search.list costs 100 quota search_request = youtube.search().list(part="id", channelId="UC_x5XG1OV2P6uZZ5FSM9Ttw", maxResults=50) # GOOD: channels.list + playlistItems.list costs 2 quota total channel_request = youtube.channels().list(part="contentDetails", id="UC_x5XG1OV2P6uZZ5FSM9Ttw") playlist_id = channel_request.execute()["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"] playlist_request = youtube.playlistItems().list(part="contentDetails", playlistId=playlist_id, maxResults=50) from redis.cluster import RedisCluster, ClusterNode nodes = [ClusterNode("redis-master-1", 6379), ClusterNode("redis-master-2", 6379)] redis_client = RedisCluster(startup_nodes=nodes, decode_responses=True) # Cache set with TTL redis_client.setex(f"yt:video:{video_id}", 3600, json.dumps(metadata)) from redis.cluster import RedisCluster, ClusterNode nodes = [ClusterNode("redis-master-1", 6379), ClusterNode("redis-master-2", 6379)] redis_client = RedisCluster(startup_nodes=nodes, decode_responses=True) # Cache set with TTL redis_client.setex(f"yt:video:{video_id}", 3600, json.dumps(metadata)) from redis.cluster import RedisCluster, ClusterNode nodes = [ClusterNode("redis-master-1", 6379), ClusterNode("redis-master-2", 6379)] redis_client = RedisCluster(startup_nodes=nodes, decode_responses=True) # Cache set with TTL redis_client.setex(f"yt:video:{video_id}", 3600, json.dumps(metadata)) from prometheus_client import Gauge QUOTA_REMAINING = Gauge("yt_api_quota_remaining", "Remaining YouTube API quota") # Update gauge after each request QUOTA_REMAINING.set(client.quota_limit - client.used_quota) from prometheus_client import Gauge QUOTA_REMAINING = Gauge("yt_api_quota_remaining", "Remaining YouTube API quota") # Update gauge after each request QUOTA_REMAINING.set(client.quota_limit - client.used_quota) from prometheus_client import Gauge QUOTA_REMAINING = Gauge("yt_api_quota_remaining", "Remaining YouTube API quota") # Update gauge after each request QUOTA_REMAINING.set(client.quota_limit - client.used_quota) youtube-api-lessons/ ├── requirements.txt # Python dependencies ├── -weight: 500;">docker-compose.yml # Local development environment ├── .env.example # Environment variable template ├── src/ │ ├── __init__.py │ ├── client.py # Core YouTubeAPIClient (Step 1) │ ├── batch_processor.py # Batch processing script (Step 2) │ ├── api.py # FastAPI REST -weight: 500;">service (Step 3) │ └── metrics.py # Prometheus metrics helpers ├── tests/ │ ├── test_client.py # Unit tests for YouTubeAPIClient │ ├── test_batch.py # Unit tests for batch processor │ └── test_api.py # Integration tests for REST API ├── data/ │ └── sample_videos.csv # Sample CSV for batch processing └── README.md # Setup and usage instructions youtube-api-lessons/ ├── requirements.txt # Python dependencies ├── -weight: 500;">docker-compose.yml # Local development environment ├── .env.example # Environment variable template ├── src/ │ ├── __init__.py │ ├── client.py # Core YouTubeAPIClient (Step 1) │ ├── batch_processor.py # Batch processing script (Step 2) │ ├── api.py # FastAPI REST -weight: 500;">service (Step 3) │ └── metrics.py # Prometheus metrics helpers ├── tests/ │ ├── test_client.py # Unit tests for YouTubeAPIClient │ ├── test_batch.py # Unit tests for batch processor │ └── test_api.py # Integration tests for REST API ├── data/ │ └── sample_videos.csv # Sample CSV for batch processing └── README.md # Setup and usage instructions youtube-api-lessons/ ├── requirements.txt # Python dependencies ├── -weight: 500;">docker-compose.yml # Local development environment ├── .env.example # Environment variable template ├── src/ │ ├── __init__.py │ ├── client.py # Core YouTubeAPIClient (Step 1) │ ├── batch_processor.py # Batch processing script (Step 2) │ ├── api.py # FastAPI REST -weight: 500;">service (Step 3) │ └── metrics.py # Prometheus metrics helpers ├── tests/ │ ├── test_client.py # Unit tests for YouTubeAPIClient │ ├── test_batch.py # Unit tests for batch processor │ └── test_api.py # Integration tests for REST API ├── data/ │ └── sample_videos.csv # Sample CSV for batch processing └── README.md # Setup and usage instructions - Agents can now create Cloudflare accounts, buy domains, and deploy (43 points) - .de TLD offline due to DNSSEC? (568 points) - Telus Uses AI to Alter Call-Agent Accents (65 points) - Accelerating Gemma 4: faster inference with multi-token prediction drafters (497 points) - StarFighter 16-Inch (98 points) - YouTube Data API v3 quota limits reset daily at midnight PST, with 10k units per project by default - google-api-python-client 2.110.0 reduces retry overhead by 40% vs 2.80.0 - Caching YouTube video metadata cuts monthly API costs by $18k for 100M daily requests - YouTube will deprecate public Data API v3 for unverified apps by Q3 2025 - Processes 1M requests/day with p99 latency under 200ms - Stays within 10% of daily API quota limits - Caches 92% of requests in Redis to avoid redundant API calls - Exposes a REST API with Prometheus metrics and health checks - Reduces monthly API costs by $18k per 100M requests - Python 3.11+ (https://www.python.org/) - Redis 7.2+ (https://redis.io/) - Google Cloud project with YouTube Data API v3 enabled (https://console.cloud.google.com/) - API key for YouTube Data API v3 (generate in Google Cloud Console) - Docker (optional, for local Redis instance) - Quota tracking prevents unexpected outages when you hit daily limits - Exponential backoff with jitter avoids thundering herd problems during outages - Redis caching reduces redundant requests for frequently accessed videos - Custom exceptions make error handling predictable across your codebase - Invalid API Key: Verify the key is enabled for YouTube Data API v3 in the Google Cloud Console. Check for typos in the key string. - Quota Exceeded Immediately: Check the "Quotas" page in Google Cloud Console to confirm your daily limit. New projects -weight: 500;">start with 10k units/day. - Redis Connection Errors: Ensure Redis is running (-weight: 500;">docker run -d -p 6379:6379 redis:7.2). Check firewall rules allow port 6379 access. - HttpError 400: Verify the video ID is valid (11 characters, alphanumeric plus - and _). - Batch Size Limit: YouTube enforces a max of 50 video IDs per videos.list call. Exceeding this returns HttpError 400. - CSV Format Issues: Ensure the CSV has a header row with a video_id column. Use UTF-8 encoding to avoid character errors. - Quota Exhaustion During Batch: Reduce batch size or split the CSV into smaller chunks if you hit quota limits mid-run. - CORS Errors: Add fastapi.middleware.cors if calling the API from a browser. Configure allowed origins explicitly. - Rate Limiting Too Strict: Adjust the @limiter.limit decorator to match your traffic patterns. - Metrics Not Showing: Ensure Prometheus is scraping the /metrics endpoint. Check that the Prometheus client is installed (-weight: 500;">pip -weight: 500;">install prometheus-client). - Team size: 4 backend engineers, 1 SRE - Stack & Versions: Python 3.11, FastAPI 0.104.0, google-api-python-client 2.110.0, Redis 7.2, PostgreSQL 16, Prometheus 2.45 - Problem: p99 latency for video metadata requests was 2.4s, daily API quota (10k) was exhausted by 11am PST, monthly API overage costs were $18k, 3 enterprise clients threatened to churn due to timeouts - Solution & Implementation: Implemented the YouTubeAPIClient with Redis caching, batch processing for nightly syncs, exponential backoff, quota tracking, and migrated expensive search.list calls to cached channels.list. Deployed the FastAPI -weight: 500;">service with rate limiting and Prometheus metrics. - Outcome: p99 latency dropped to 120ms, quota lasted full 24h cycle, overage costs eliminated (saving $18k/month), churn risk resolved, cache hit rate reached 92% - With YouTube’s planned deprecation of public Data API v3 for unverified apps in Q3 2025, how will your team adapt to the new OAuth-only requirements? - Is the 75% latency reduction from Redis caching worth the 12% increase in infrastructure costs for your use case? - How does the youtube-dl fork yt-dlp compare to the official API for metadata extraction, and when would you choose one over the other?