DEV Community: ahmet gedik

High-Performance Video API Server with Go net/http

ahmet gedik — Sun, 12 Apr 2026 12:00:01 +0000

High-Performance Video API Server with Go net/http

Go is compelling for a video platform API: compiled binaries, zero-overhead goroutines, and a production-ready HTTP server in the standard library. Here's how I'd build the API backend for TopVideoHub in Go.

Why Go for This Use Case

net/http handles ~100k requests/second on modest hardware
Goroutines cost ~2KB memory vs ~1MB for OS threads
Single binary deployment — no runtime, no dependencies
Built-in race detector during development

For TopVideoHub handling 9 Asia-Pacific regions with concurrent trending requests, Go's concurrency model is a natural fit.

Core Types

package models

import "time"

type Region string

const (
    JP Region = "JP"
    KR Region = "KR"
    TW Region = "TW"
    SG Region = "SG"
    VN Region = "VN"
    TH Region = "TH"
    HK Region = "HK"
    US Region = "US"
    GB Region = "GB"
)

var ValidRegions = map[Region]bool{
    JP: true, KR: true, TW: true, SG: true,
    VN: true, TH: true, HK: true, US: true, GB: true,
}

// Video holds all metadata for a single video entry.
// Title, ChannelTitle may contain CJK characters — Go strings are UTF-8 natively.
type Video struct {
    VideoID      string    `json:"video_id"`
    Title        string    `json:"title"`
    ChannelTitle string    `json:"channel_title"`
    ThumbnailURL string    `json:"thumbnail_url"`
    ViewCount    int64     `json:"view_count"`
    Region       Region    `json:"region"`
    CategoryID   int       `json:"category_id"`
    PublishedAt  time.Time `json:"published_at"`
}

type TrendingResponse struct {
    Region    Region    `json:"region"`
    Videos    []Video   `json:"videos"`
    Total     int       `json:"total"`
    FetchedAt time.Time `json:"fetched_at"`
}

HTTP Handler

package handlers

import (
    "encoding/json"
    "net/http"
    "strings"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
    "topvideohub/internal/models"
)

type TrendingHandler struct {
    Pool  *pgxpool.Pool
    Cache *Cache
}

func (h *TrendingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // Extract region: /trending/JP
    parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
    if len(parts) < 2 || parts[1] == "" {
        http.Error(w, `{"error":"region required"}`, http.StatusBadRequest)
        return
    }

    region := models.Region(strings.ToUpper(parts[1]))
    if !models.ValidRegions[region] {
        http.Error(w, `{"error":"invalid region"}`, http.StatusNotFound)
        return
    }

    cacheKey := "trending:" + string(region)
    if cached, ok := h.Cache.Get(cacheKey); ok {
        w.Header().Set("Content-Type", "application/json")
        w.Header().Set("X-Cache", "HIT")
        w.Write(cached)
        return
    }

    rows, err := h.Pool.Query(r.Context(), `
        SELECT v.video_id, v.title, v.channel_title, v.thumbnail_url,
               v.view_count, v.category_id, v.published_at, vr.region
        FROM videos v
        JOIN video_regions vr ON vr.video_id = v.id
        WHERE vr.region = $1
        ORDER BY vr.trending_rank ASC NULLS LAST, v.view_count DESC
        LIMIT 50
    `, string(region))
    if err != nil {
        http.Error(w, `{"error":"database error"}`, http.StatusInternalServerError)
        return
    }
    defer rows.Close()

    var videos []models.Video
    for rows.Next() {
        var v models.Video
        if err := rows.Scan(
            &v.VideoID, &v.Title, &v.ChannelTitle, &v.ThumbnailURL,
            &v.ViewCount, &v.CategoryID, &v.PublishedAt, &v.Region,
        ); err != nil {
            continue
        }
        videos = append(videos, v)
    }

    resp := models.TrendingResponse{
        Region:    region,
        Videos:    videos,
        Total:     len(videos),
        FetchedAt: time.Now().UTC(),
    }

    data, _ := json.Marshal(resp)
    h.Cache.Set(cacheKey, data, time.Hour)

    w.Header().Set("Content-Type", "application/json")
    w.Header().Set("Cache-Control", "public, max-age=3600")
    w.Header().Set("X-Cache", "MISS")
    w.Write(data)
}

Middleware Chaining

package middleware

import (
    "log"
    "net/http"
    "time"
)

// Logger logs each request with method, path, status, and duration.
func Logger(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()
        rw := &responseWriter{ResponseWriter: w, status: http.StatusOK}
        next.ServeHTTP(rw, r)
        log.Printf("%s %s %d %v", r.Method, r.URL.Path, rw.status, time.Since(start))
    })
}

// CORS adds permissive CORS headers for API access.
func CORS(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Access-Control-Allow-Origin", "*")
        w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
        if r.Method == http.MethodOptions {
            w.WriteHeader(http.StatusNoContent)
            return
        }
        next.ServeHTTP(w, r)
    })
}

Main Server

func main() {
    ctx := context.Background()
    pool, _ := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
    cache := handlers.NewCache()

    mux := http.NewServeMux()
    mux.Handle("/trending/", &handlers.TrendingHandler{Pool: pool, Cache: cache})
    mux.Handle("/search", &handlers.SearchHandler{Pool: pool, Cache: cache})
    mux.Handle("/categories", &handlers.CategoriesHandler{Pool: pool, Cache: cache})

    handler := middleware.Logger(middleware.CORS(mux))

    srv := &http.Server{
        Addr:         ":8080",
        Handler:      handler,
        ReadTimeout:  5 * time.Second,
        WriteTimeout: 10 * time.Second,
        IdleTimeout:  60 * time.Second,
    }
    log.Printf("TopVideoHub Go API serving on :8080")
    log.Fatal(srv.ListenAndServe())
}

This Go API for TopVideoHub handles concurrent requests for all 9 Asia-Pacific regions efficiently. The in-memory cache layer prevents repeated database hits for trending data that changes only hourly. Go strings are UTF-8 natively, so Korean, Japanese, and Chinese video titles are serialized correctly to JSON without any special handling.

This article is part of the Building TopVideoHub series. Check out TopVideoHub to see these techniques in action.

Video Platform API with Python FastAPI

ahmet gedik — Sun, 12 Apr 2026 00:00:01 +0000

Building a Video API from Scratch

Every video platform needs an API — whether it serves a frontend SPA, a mobile app, or third-party integrations. At DailyWatch, our API powers video search, category browsing, and trending feeds across 8 regions. Here is how to build one with FastAPI.

Project Structure

video_api/
├── main.py
├── models.py
├── database.py
├── routers/
│   ├── videos.py
│   ├── categories.py
│   └── trending.py

Pydantic Models

Define your data shapes once, get validation and serialization for free:

from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum

class Region(str, Enum):
    US = "US"
    GB = "GB"
    DE = "DE"
    FR = "FR"
    IN_ = "IN"
    BR = "BR"
    AU = "AU"
    CA = "CA"

class VideoBase(BaseModel):
    video_id: str
    title: str
    channel_title: str
    category_id: int
    thumbnail_url: str
    views: int = 0
    likes: int = 0
    duration: int = 0
    regions: list[str] = []

class VideoDetail(VideoBase):
    description: str = ""
    published_at: datetime | None = None
    fetched_at: datetime

class VideoSearchResult(BaseModel):
    videos: list[VideoBase]
    total: int
    page: int
    per_page: int
    query: str

class TrendingResponse(BaseModel):
    region: str
    videos: list[VideoBase]
    updated_at: datetime

Pydantic validates incoming requests and serializes outgoing responses. The Region enum restricts region parameters to valid values automatically.

Async Database Layer

FastAPI is async-native, so our database layer should be too:

import aiosqlite
from contextlib import asynccontextmanager

DATABASE_PATH = "data/videos.db"

@asynccontextmanager
async def get_db():
    db = await aiosqlite.connect(DATABASE_PATH)
    db.row_factory = aiosqlite.Row
    await db.execute("PRAGMA journal_mode=WAL")
    await db.execute("PRAGMA cache_size=-8000")
    try:
        yield db
    finally:
        await db.close()

async def search_videos(query: str, page: int = 1, per_page: int = 20) -> tuple[list[dict], int]:
    async with get_db() as db:
        # Count total matches
        cursor = await db.execute(
            "SELECT COUNT(*) FROM videos_fts WHERE videos_fts MATCH ?",
            (query,)
        )
        row = await cursor.fetchone()
        total = row[0]

        # Fetch page of results
        offset = (page - 1) * per_page
        cursor = await db.execute("""
            SELECT v.video_id, v.title, v.channel_title, v.category_id,
                   v.thumbnail_url, v.views, v.likes, v.duration,
                   GROUP_CONCAT(vr.region) as regions
            FROM videos v
            JOIN videos_fts ON v.video_id = videos_fts.video_id
            LEFT JOIN video_regions vr ON v.video_id = vr.video_id
            WHERE videos_fts MATCH ?
            GROUP BY v.video_id
            ORDER BY rank
            LIMIT ? OFFSET ?
        """, (query, per_page, offset))
        rows = await cursor.fetchall()
        videos = [dict(row) for row in rows]
        for v in videos:
            v["regions"] = v["regions"].split(",") if v["regions"] else []
        return videos, total

Using aiosqlite wraps SQLite in an async interface so database queries do not block the event loop.

API Endpoints

from fastapi import FastAPI, Query, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="DailyWatch API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dailywatch.video"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

@app.get("/api/search", response_model=VideoSearchResult)
async def search(
    q: str = Query(..., min_length=2, max_length=100, description="Search query"),
    page: int = Query(1, ge=1, le=100),
    per_page: int = Query(20, ge=1, le=50),
):
    videos, total = await search_videos(q, page, per_page)
    return VideoSearchResult(
        videos=videos,
        total=total,
        page=page,
        per_page=per_page,
        query=q,
    )

@app.get("/api/trending/{region}", response_model=TrendingResponse)
async def trending(region: Region):
    async with get_db() as db:
        cursor = await db.execute("""
            SELECT v.video_id, v.title, v.channel_title, v.category_id,
                   v.thumbnail_url, v.views, v.likes, v.duration
            FROM videos v
            JOIN video_regions vr ON v.video_id = vr.video_id
            WHERE vr.region = ?
            ORDER BY v.views DESC
            LIMIT 50
        """, (region.value,))
        rows = await cursor.fetchall()
        return TrendingResponse(
            region=region.value,
            videos=[dict(r) for r in rows],
            updated_at=datetime.utcnow(),
        )

@app.get("/api/categories")
async def list_categories():
    async with get_db() as db:
        cursor = await db.execute(
            "SELECT id, title FROM categories ORDER BY title"
        )
        return [dict(r) for r in await cursor.fetchall()]

Dependency Injection for Caching

FastAPI's dependency injection makes adding a cache layer clean:

from functools import lru_cache
from datetime import datetime, timedelta

_cache: dict[str, tuple[datetime, any]] = {}

async def cached_categories() -> list[dict]:
    key = "categories"
    if key in _cache:
        expires, data = _cache[key]
        if datetime.utcnow() < expires:
            return data

    async with get_db() as db:
        cursor = await db.execute("SELECT id, title FROM categories ORDER BY title")
        data = [dict(r) for r in await cursor.fetchall()]

    _cache[key] = (datetime.utcnow() + timedelta(hours=24), data)
    return data

@app.get("/api/categories")
async def list_categories(categories: list = Depends(cached_categories)):
    return categories

Categories rarely change, so caching them for 24 hours eliminates a database query on every request.

Running and Testing

pip install fastapi uvicorn aiosqlite
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

FastAPI auto-generates OpenAPI docs at /docs. For DailyWatch, this API serves as the backbone for both server-rendered pages and any future mobile or SPA frontend.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

Async Video Pipeline with Python asyncio for Multi-Region Fetching

ahmet gedik — Sat, 11 Apr 2026 16:00:06 +0000

Async Video Pipeline with Python asyncio for Multi-Region Fetching

TrendVidStream fetches trending video data from 8 regions: US, GB, AE (UAE), FI (Finland), DK (Denmark), CZ (Czech Republic), BE (Belgium), and CH (Switzerland). Sequential fetching is slow; asyncio makes it fast.

Here's the async pipeline I built, including the quirks specific to a diverse multi-region set.

Core Fetch Function

import asyncio
import aiohttp
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger(__name__)

REGIONS = ["AE", "FI", "DK", "CZ", "BE", "CH", "US", "GB"]
YT_BASE = "https://www.googleapis.com/youtube/v3"


@dataclass
class VideoMetadata:
    video_id: str
    title: str
    channel_title: str
    region: str
    view_count: int
    thumbnail_url: str
    published_at: str
    category_id: int
    language: Optional[str] = None  # defaultAudioLanguage from snippet


async def fetch_trending(
    session: aiohttp.ClientSession,
    region: str,
    api_key: str,
    sem: asyncio.Semaphore,
) -> list[VideoMetadata]:
    """Fetch trending videos for one region with backpressure control."""
    async with sem:
        params = {
            "part": "snippet,statistics",
            "chart": "mostPopular",
            "regionCode": region,
            "maxResults": 50,
            "key": api_key,
        }
        try:
            async with session.get(
                f"{YT_BASE}/videos",
                params=params,
                timeout=aiohttp.ClientTimeout(total=30),
            ) as resp:
                if resp.status == 403:
                    logger.error(f"[{region}] 403 Forbidden — check API key quota")
                    return []
                if resp.status == 429:
                    logger.warning(f"[{region}] Rate limited — sleeping 5s")
                    await asyncio.sleep(5)
                    return []  # Will be retried on next cron cycle
                resp.raise_for_status()
                data = await resp.json()
                return [_parse(item, region) for item in data.get("items", [])]
        except asyncio.TimeoutError:
            logger.error(f"[{region}] Timeout")
            return []
        except aiohttp.ClientError as exc:
            logger.error(f"[{region}] HTTP error: {exc}")
            return []


def _parse(item: dict, region: str) -> VideoMetadata:
    s = item["snippet"]
    stats = item.get("statistics", {})
    thumbs = s.get("thumbnails", {})
    thumb = next(
        (thumbs[q]["url"] for q in ("maxres", "standard", "high", "medium", "default") if q in thumbs),
        "",
    )
    return VideoMetadata(
        video_id=item["id"],
        title=s.get("title", ""),
        channel_title=s.get("channelTitle", ""),
        region=region,
        view_count=int(stats.get("viewCount", 0)),
        thumbnail_url=thumb,
        published_at=s.get("publishedAt", ""),
        category_id=int(s.get("categoryId", 0)),
        language=s.get("defaultAudioLanguage"),  # 'ar' for UAE, 'fi' for Finland, etc.
    )

Concurrent Fetch for All Regions

async def fetch_all(
    api_key: str,
    regions: list[str] = REGIONS,
    concurrency: int = 4,
) -> dict[str, list[VideoMetadata]]:
    """Fetch all regions with bounded concurrency."""
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(limit=8, ttl_dns_cache=300)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            asyncio.create_task(fetch_trending(session, r, api_key, sem))
            for r in regions
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return {
        region: (result if not isinstance(result, Exception) else [])
        for region, result in zip(regions, results)
    }

Handling Arabic Text from UAE

Arabic titles from the UAE feed come through as UTF-8 — Python handles this naturally. But there's a subtlety: Arabic text includes Unicode bidirectional control characters that can cause display issues if stored raw:

import unicodedata
import re

RTL_OVERRIDES = str.maketrans('', '', '\u200f\u200e\u202a\u202b\u202c\u202d\u202e')

def clean_title(title: str) -> str:
    """Remove BiDi control characters, normalize to NFC."""
    title = title.translate(RTL_OVERRIDES)  # Remove BiDi overrides
    title = unicodedata.normalize('NFC', title)  # Canonical composition
    return title.strip()

Processing Pipeline with Language Detection

import asyncpg
from langdetect import detect, LangDetectException  # pip install langdetect

async def process_and_store(
    pool: asyncpg.Pool,
    all_videos: dict[str, list[VideoMetadata]]
) -> dict[str, int]:
    """Process videos and bulk-upsert to PostgreSQL."""
    processed = {}

    for region, videos in all_videos.items():
        clean_videos = []
        for v in videos:
            # Clean titles (especially important for Arabic RTL markers)
            v.title = clean_title(v.title)

            # If API doesn't provide language, detect it
            if not v.language and v.title:
                try:
                    v.language = detect(v.title)
                except LangDetectException:
                    v.language = 'und'  # Undetermined

            clean_videos.append(v)

        # Bulk upsert to PostgreSQL
        async with pool.acquire() as conn:
            await conn.executemany("""
                INSERT INTO videos
                    (video_id, title, channel_title, region, view_count,
                     thumbnail_url, published_at, category_id, language)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                ON CONFLICT (video_id, region) DO UPDATE SET
                    view_count = EXCLUDED.view_count,
                    fetched_at = now()
            """, [
                (v.video_id, v.title, v.channel_title, v.region,
                 v.view_count, v.thumbnail_url, v.published_at,
                 v.category_id, v.language)
                for v in clean_videos
            ])
        processed[region] = len(clean_videos)
        logger.info(f"[{region}] Stored {len(clean_videos)} videos")

    return processed


async def main():
    import os
    api_key = os.environ["YOUTUBE_API_KEY"]
    db_url = os.environ["DATABASE_URL"]

    pool = await asyncpg.create_pool(db_url, min_size=2, max_size=5)
    all_videos = await fetch_all(api_key)
    result = await process_and_store(pool, all_videos)

    total = sum(result.values())
    logger.info(f"Pipeline complete: {total} videos across {len(result)} regions")
    await pool.close()

For TrendVidStream's 8 regions, this async pipeline runs in ~3 seconds vs ~16 seconds sequential. The defaultAudioLanguage field from YouTube's API is invaluable for UAE content — it tells you whether a video is primarily Arabic, Hindi, English, or Urdu, which drives the language mix analytics.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.

Async Video Processing Pipeline with Python for European Content

ahmet gedik — Sat, 11 Apr 2026 07:00:01 +0000

Processing Video Data from 7 European Regions

ViralVidVault fetches trending videos from Poland, Netherlands, Sweden, Norway, Austria, UK, and US every 7 hours. Each API call has 200-500ms of network latency. Done sequentially, that is 1.4-3.5 seconds of waiting. With Python's asyncio, all 7 fetches run concurrently and the total time drops to one round-trip.

The Fetcher with Regional Awareness

European APIs sometimes have quirks — rate limit differences, occasional regional outages, content encoding issues with non-ASCII characters. The fetcher accounts for these:

import asyncio
import aiohttp
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class RegionalVideo:
    video_id: str
    title: str
    channel: str
    views: int
    region: str
    language: str = "en"
    fetched_at: datetime = field(default_factory=datetime.utcnow)

REGION_LANGUAGES = {
    "PL": "pl", "NL": "nl", "SE": "sv",
    "NO": "no", "AT": "de", "GB": "en", "US": "en",
}

class EuropeanFetcher:
    def __init__(self, api_key: str, max_concurrent: int = 3):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.stats = {"fetched": 0, "errors": 0, "regions_ok": []}

    async def fetch_region(self, session: aiohttp.ClientSession, region: str) -> list[RegionalVideo]:
        async with self.semaphore:
            url = "https://www.googleapis.com/youtube/v3/videos"
            params = {
                "part": "snippet,statistics,contentDetails",
                "chart": "mostPopular",
                "regionCode": region,
                "maxResults": 25,
                "key": self.api_key,
            }
            try:
                async with session.get(url, params=params,
                                       timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 429:
                        wait = int(resp.headers.get("Retry-After", 5))
                        print(f"[{region}] Rate limited, waiting {wait}s")
                        await asyncio.sleep(wait)
                        return await self.fetch_region(session, region)

                    resp.raise_for_status()
                    data = await resp.json()
                    lang = REGION_LANGUAGES.get(region, "en")

                    videos = [
                        RegionalVideo(
                            video_id=item["id"],
                            title=item["snippet"]["title"],
                            channel=item["snippet"].get("channelTitle", ""),
                            views=int(item["statistics"].get("viewCount", 0)),
                            region=region,
                            language=lang,
                        )
                        for item in data.get("items", [])
                    ]
                    self.stats["fetched"] += len(videos)
                    self.stats["regions_ok"].append(region)
                    return videos

            except asyncio.TimeoutError:
                print(f"[{region}] Timeout")
                self.stats["errors"] += 1
                return []
            except aiohttp.ClientError as e:
                print(f"[{region}] Error: {e}")
                self.stats["errors"] += 1
                return []

    async def fetch_all(self) -> dict[str, list[RegionalVideo]]:
        regions = ["US", "GB", "PL", "NL", "SE", "NO", "AT"]
        async with aiohttp.ClientSession() as session:
            tasks = {r: self.fetch_region(session, r) for r in regions}
            results = {}
            gathered = await asyncio.gather(
                *[self.fetch_region(session, r) for r in regions],
                return_exceptions=True,
            )
            for region, result in zip(regions, gathered):
                if isinstance(result, Exception):
                    print(f"[{region}] Exception: {result}")
                    results[region] = []
                else:
                    results[region] = result
            return results

The REGION_LANGUAGES mapping is essential for downstream processing — it tells the search indexer which PostgreSQL dictionary to use and helps with content categorization.

Deduplication with Region Merging

A viral music video might trend in Poland, Netherlands, and Sweden simultaneously:

def deduplicate(region_results: dict[str, list[RegionalVideo]]) -> list[dict]:
    seen: dict[str, dict] = {}

    for region, videos in region_results.items():
        for video in videos:
            if video.video_id in seen:
                seen[video.video_id]["regions"].append(region)
                seen[video.video_id]["views"] = max(
                    seen[video.video_id]["views"], video.views
                )
            else:
                seen[video.video_id] = {
                    "video_id": video.video_id,
                    "title": video.title,
                    "channel": video.channel,
                    "views": video.views,
                    "regions": [region],
                    "primary_language": video.language,
                }

    return list(seen.values())

For ViralVidVault, cross-region overlap is less than 5% among European regions — much lower than you might expect. This means most videos appear in exactly one region's trending feed.

Running the Full Pipeline

async def run_pipeline(api_key: str):
    fetcher = EuropeanFetcher(api_key, max_concurrent=3)

    # Stage 1: Fetch all regions concurrently
    raw = await fetcher.fetch_all()
    total_raw = sum(len(v) for v in raw.values())
    print(f"Fetched {total_raw} videos from {len(fetcher.stats['regions_ok'])} regions")

    # Stage 2: Deduplicate
    unique = deduplicate(raw)
    print(f"After dedup: {len(unique)} unique videos")

    # Stage 3: Store (using your DB layer)
    # await store_videos(unique)

    return {"raw": total_raw, "unique": len(unique), "stats": fetcher.stats}

if __name__ == "__main__":
    import os
    result = asyncio.run(run_pipeline(os.environ["YOUTUBE_API_KEY"]))
    print(f"Pipeline complete: {result}")

Performance Numbers

On a typical 7-region fetch for ViralVidVault:

Approach	Total Time
Sequential for loop	2.8s
asyncio.gather (unlimited)	0.45s
asyncio + Semaphore(3)	0.7s

The semaphore version is 4x faster than sequential while respecting rate limits. The full pipeline (fetch + dedup + store) completes in about 1.5 seconds.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

Building a Video Platform API with Python FastAPI

ahmet gedik — Fri, 10 Apr 2026 20:00:12 +0000

Building a Video Platform API with Python FastAPI

FastAPI is an excellent match for a video platform API — async by design, automatic OpenAPI docs, and Pydantic validation that handles multi-script video titles cleanly. Here's how I'd build the API layer for TopVideoHub.

Project Setup

pip install fastapi uvicorn asyncpg pydantic

Pydantic Models

Pydantic validates input and serializes output automatically. Importantly, it handles CJK Unicode natively — no special configuration needed for Japanese or Korean titles:

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum


class Region(str, Enum):
    JP = "JP"
    KR = "KR"
    TW = "TW"
    SG = "SG"
    VN = "VN"
    TH = "TH"
    HK = "HK"
    US = "US"
    GB = "GB"


class VideoResponse(BaseModel):
    video_id: str
    title: str  # May contain Japanese, Korean, Chinese, etc.
    channel_title: str
    thumbnail_url: str
    view_count: int
    published_at: datetime
    region: Region
    category_id: int
    duration: Optional[str] = None

    class Config:
        json_encoders = {datetime: lambda v: v.isoformat()}


class SearchResponse(BaseModel):
    query: str
    region: Optional[Region]
    results: list[VideoResponse]
    total: int
    page: int
    per_page: int


class TrendingResponse(BaseModel):
    region: Region
    videos: list[VideoResponse]
    fetched_at: datetime
    total: int

Dependency Injection for Database

FastAPI's dependency system cleanly manages database connection lifecycles:

# database.py
import asyncpg
from typing import AsyncGenerator

_pool: asyncpg.Pool | None = None


async def init_pool(dsn: str) -> None:
    global _pool
    _pool = await asyncpg.create_pool(
        dsn,
        min_size=2,
        max_size=10,
        command_timeout=30,
    )


async def get_conn() -> AsyncGenerator[asyncpg.Connection, None]:
    """FastAPI dependency: yields a DB connection per request."""
    async with _pool.acquire() as conn:
        yield conn

Search Endpoint with PostgreSQL FTS

@router.get("/search", response_model=SearchResponse)
async def search_videos(
    q: str = Query(..., min_length=1, max_length=200),
    region: Region | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    per_page: int = Query(default=20, ge=1, le=50),
    conn: asyncpg.Connection = Depends(get_conn),
):
    offset = (page - 1) * per_page

    # plainto_tsquery handles CJK terms gracefully via 'simple' config
    base_query = """
        SELECT v.video_id, v.title, v.channel_title, v.thumbnail_url,
               v.view_count, v.published_at, v.category_id, v.region,
               ts_rank(v.search_vec,
                   to_tsquery('simple', plainto_tsquery('simple', $1)::text),
                   32) AS rank
        FROM videos v
        WHERE v.search_vec @@ to_tsquery('simple', plainto_tsquery('simple', $1)::text)
    """
    params: list = [q]

    if region:
        base_query += " AND v.region = $2"
        params.append(region.value)

    base_query += f" ORDER BY rank DESC LIMIT ${len(params)+1} OFFSET ${len(params)+2}"
    params.extend([per_page, offset])

    rows = await conn.fetch(base_query, *params)
    videos = [VideoResponse(**dict(r)) for r in rows]

    return SearchResponse(
        query=q,
        region=region,
        results=videos,
        total=len(videos),
        page=page,
        per_page=per_page,
    )

Application Entry Point

from fastapi import FastAPI
from contextlib import asynccontextmanager
import os


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_pool(os.environ["DATABASE_URL"])
    yield
    if _pool:
        await _pool.close()


app = FastAPI(
    title="TopVideoHub API",
    description="REST API for trending video data across 9 Asia-Pacific regions",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(trending_router)
app.include_router(search_router)
app.include_router(categories_router)

FastAPI's automatic /docs endpoint generates interactive Swagger UI — invaluable for exploring multi-region queries across TopVideoHub's video data. The Pydantic Region enum means FastAPI validates region codes automatically and returns a clear 422 error for invalid inputs, rather than letting bad data reach the database.

This article is part of the Building TopVideoHub series. Check out TopVideoHub to see these techniques in action.

Async Video Processing Pipeline with Python asyncio

ahmet gedik — Fri, 10 Apr 2026 12:00:01 +0000

Why Async Matters for Video Metadata

When DailyWatch fetches trending videos from 8 regions, each API call takes 200-500ms due to network latency. Done sequentially, that is 1.6-4 seconds just for the fetches. With asyncio, all 8 requests fire simultaneously and complete in the time of the slowest single request.

The Basic Pattern

Here is the simplest async fetch pattern:

import asyncio
import aiohttp
from dataclasses import dataclass

@dataclass
class VideoResult:
    video_id: str
    title: str
    views: int
    region: str

async def fetch_trending(session: aiohttp.ClientSession, region: str, api_key: str) -> list[VideoResult]:
    url = "https://www.googleapis.com/youtube/v3/videos"
    params = {
        "part": "snippet,statistics",
        "chart": "mostPopular",
        "regionCode": region,
        "maxResults": 50,
        "key": api_key,
    }
    async with session.get(url, params=params) as resp:
        data = await resp.json()
        return [
            VideoResult(
                video_id=item["id"],
                title=item["snippet"]["title"],
                views=int(item["statistics"].get("viewCount", 0)),
                region=region,
            )
            for item in data.get("items", [])
        ]

async def main():
    regions = ["US", "GB", "DE", "FR", "IN", "BR", "AU", "CA"]
    api_key = "YOUR_API_KEY"

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_trending(session, r, api_key) for r in regions]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for region, result in zip(regions, results):
        if isinstance(result, Exception):
            print(f"[{region}] Error: {result}")
        else:
            print(f"[{region}] Fetched {len(result)} videos")

asyncio.run(main())

asyncio.gather runs all tasks concurrently. The return_exceptions=True flag prevents one failed region from killing the entire batch.

Adding Rate Limiting with a Semaphore

APIs have rate limits. A semaphore controls how many requests run simultaneously:

import asyncio
import aiohttp
from asyncio import Semaphore

class RateLimitedFetcher:
    def __init__(self, api_key: str, max_concurrent: int = 3):
        self.api_key = api_key
        self.semaphore = Semaphore(max_concurrent)
        self.request_count = 0

    async def fetch_region(self, session: aiohttp.ClientSession, region: str) -> list[dict]:
        async with self.semaphore:
            self.request_count += 1
            url = "https://www.googleapis.com/youtube/v3/videos"
            params = {
                "part": "snippet,statistics,contentDetails",
                "chart": "mostPopular",
                "regionCode": region,
                "maxResults": 50,
                "key": self.api_key,
            }
            try:
                async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 429:
                        retry_after = int(resp.headers.get("Retry-After", 5))
                        print(f"[{region}] Rate limited, waiting {retry_after}s")
                        await asyncio.sleep(retry_after)
                        return await self.fetch_region(session, region)
                    resp.raise_for_status()
                    data = await resp.json()
                    return data.get("items", [])
            except asyncio.TimeoutError:
                print(f"[{region}] Timeout after 10s")
                return []
            except aiohttp.ClientError as e:
                print(f"[{region}] HTTP error: {e}")
                return []

    async def fetch_all_regions(self, regions: list[str]) -> dict[str, list]:
        async with aiohttp.ClientSession() as session:
            tasks = {r: self.fetch_region(session, r) for r in regions}
            results = {}
            for region, coro in tasks.items():
                results[region] = await coro
            return results

With max_concurrent=3, at most 3 API calls run at once. The 429 handler automatically retries after the server-specified delay.

Building a Processing Pipeline

Fetching is only step one. We also need to normalize, deduplicate, and store. Here is a pipeline with separate stages:

import asyncio
from collections import defaultdict

async def normalize(raw_items: list[dict], region: str) -> list[dict]:
    """Transform API response into our internal format."""
    normalized = []
    for item in raw_items:
        snippet = item.get("snippet", {})
        stats = item.get("statistics", {})
        normalized.append({
            "video_id": item["id"],
            "title": snippet.get("title", ""),
            "channel": snippet.get("channelTitle", ""),
            "views": int(stats.get("viewCount", 0)),
            "likes": int(stats.get("likeCount", 0)),
            "region": region,
            "category_id": int(snippet.get("categoryId", 0)),
        })
    return normalized

async def deduplicate(all_videos: list[dict]) -> list[dict]:
    """Merge duplicate videos from different regions."""
    seen: dict[str, dict] = {}
    for video in all_videos:
        vid = video["video_id"]
        if vid in seen:
            seen[vid]["regions"].append(video["region"])
            seen[vid]["views"] = max(seen[vid]["views"], video["views"])
        else:
            video["regions"] = [video["region"]]
            seen[vid] = video
    return list(seen.values())

async def run_pipeline(api_key: str):
    regions = ["US", "GB", "DE", "FR", "IN", "BR", "AU", "CA"]
    fetcher = RateLimitedFetcher(api_key, max_concurrent=3)

    # Stage 1: Fetch
    raw_results = await fetcher.fetch_all_regions(regions)
    print(f"Fetched from {len(raw_results)} regions, {fetcher.request_count} API calls")

    # Stage 2: Normalize (parallel per region)
    normalize_tasks = [
        normalize(items, region) for region, items in raw_results.items()
    ]
    normalized_batches = await asyncio.gather(*normalize_tasks)

    # Stage 3: Flatten and deduplicate
    all_videos = [v for batch in normalized_batches for v in batch]
    unique_videos = await deduplicate(all_videos)
    print(f"Total: {len(all_videos)} raw, {len(unique_videos)} after dedup")

    return unique_videos

Each stage is async-compatible. The normalization runs in parallel across regions with asyncio.gather, and the deduplication merges region tags for videos trending in multiple countries.

Error Handling Patterns

In production at DailyWatch, we wrap the entire pipeline with structured error handling:

async def safe_pipeline(api_key: str) -> dict:
    stats = {"fetched": 0, "errors": [], "deduplicated": 0}
    try:
        videos = await run_pipeline(api_key)
        stats["fetched"] = len(videos)
        stats["deduplicated"] = len(videos)
    except Exception as e:
        stats["errors"].append(str(e))
    return stats

The key insight is using return_exceptions=True in gather calls so one region failure does not abort the others, and logging errors per-region for debugging.

Performance Comparison

On an 8-region fetch for DailyWatch:

Approach	Time
Sequential (for loop)	3.2s
asyncio.gather (unlimited)	0.5s
asyncio + semaphore(3)	0.9s

The semaphore approach is the sweet spot — 3.5x faster than sequential while respecting rate limits.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

PostgreSQL JSONB for Multi-Region Video Analytics

ahmet gedik — Fri, 10 Apr 2026 03:00:01 +0000

PostgreSQL JSONB for Multi-Region Video Analytics

TrendVidStream serves video content from 8 regions across vastly different markets — the UAE's multilingual melting pot, Finland's education-heavy viewership, and Czech Republic's thriving local creator scene. Tracking analytics for these diverse markets requires a schema that adapts to new metrics without constant migrations.

PostgreSQL's JSONB is the answer.

Schema Design

CREATE TABLE video_analytics (
    id          BIGSERIAL PRIMARY KEY,
    video_id    TEXT NOT NULL,
    date        DATE NOT NULL,
    region      TEXT NOT NULL,  -- AE, FI, DK, CZ, BE, CH, US, GB
    total_views INTEGER DEFAULT 0,
    -- Flexible analytics payload
    metrics     JSONB DEFAULT '{}'::jsonb,
    updated_at  TIMESTAMPTZ DEFAULT now(),
    UNIQUE(video_id, date, region)
);

-- Region+date is the most common access pattern
CREATE INDEX idx_analytics_region_date
    ON video_analytics(region, date DESC);

-- GIN index enables efficient @> containment and jsonb_path queries
CREATE INDEX idx_analytics_metrics_gin
    ON video_analytics USING GIN(metrics jsonb_path_ops);

-- Expression indexes for frequently-queried numeric fields
CREATE INDEX idx_analytics_completion
    ON video_analytics (((metrics->>'completion_rate')::real))
    WHERE metrics ? 'completion_rate';

CREATE INDEX idx_analytics_mobile
    ON video_analytics (((metrics->'device'->>'mobile')::real))
    WHERE metrics->'device' IS NOT NULL;

Inserting Rich Analytics

Each region has different analytics characteristics. UAE has high mobile usage and multilingual engagement, Nordic regions show higher completion rates, Czech content stays mostly local:

INSERT INTO video_analytics (video_id, date, region, total_views, metrics)
VALUES (
    'abc123',
    CURRENT_DATE,
    'AE',
    42000,
    '{
        "completion_rate": 0.54,
        "avg_watch_seconds": 143,
        "device": {"mobile": 0.83, "desktop": 0.14, "tv": 0.03},
        "language_mix": {"ar": 0.48, "en": 0.28, "hi": 0.15, "ur": 0.09},
        "cross_region_trending": ["AE", "SA", "EG"],
        "peak_hour_utc": 18
    }'::jsonb
)
ON CONFLICT (video_id, date, region) DO UPDATE SET
    total_views = video_analytics.total_views + EXCLUDED.total_views,
    metrics = video_analytics.metrics || EXCLUDED.metrics,
    updated_at = now();

Analytics Queries for Platform Insights

-- Compare mobile usage across TrendVidStream regions
SELECT
    region,
    SUM(total_views) AS total_views,
    ROUND(AVG((metrics->'device'->>'mobile')::real)::numeric, 3) AS avg_mobile,
    ROUND(AVG((metrics->>'completion_rate')::real)::numeric, 3) AS avg_completion
FROM video_analytics
WHERE date >= CURRENT_DATE - interval '30 days'
GROUP BY region
ORDER BY avg_mobile DESC;

This query reveals interesting patterns: UAE's 83% mobile viewership vs Finland's ~62% reflects infrastructure and device preferences across markets.

-- Find content that crosses borders in the Middle East/Nordic markets
SELECT
    video_id,
    array_agg(region ORDER BY total_views DESC) AS regions,
    sum(total_views) AS global_views
FROM video_analytics
WHERE date = CURRENT_DATE
  AND metrics->'cross_region_trending' IS NOT NULL
GROUP BY video_id
HAVING count(*) > 2
ORDER BY global_views DESC
LIMIT 20;

Querying JSONB Arrays

For language mix analysis specific to the UAE feed on TrendVidStream:

-- Videos with significant Arabic-language engagement in UAE
SELECT
    va.video_id,
    v.title,
    va.total_views,
    (va.metrics->'language_mix'->>'ar')::real AS arabic_ratio
FROM video_analytics va
JOIN videos v ON v.video_id = va.video_id
WHERE va.region = 'AE'
  AND va.date >= CURRENT_DATE - interval '7 days'
  AND (va.metrics->'language_mix'->>'ar')::real > 0.4
ORDER BY va.total_views DESC
LIMIT 10;

Python Integration

import asyncpg
import json
from datetime import date
from typing import Any

async def record_regional_analytics(
    pool: asyncpg.Pool,
    video_id: str,
    region: str,
    total_views: int,
    metrics: dict[str, Any]
) -> None:
    async with pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO video_analytics
                (video_id, date, region, total_views, metrics)
            VALUES ($1, $2, $3, $4, $5::jsonb)
            ON CONFLICT (video_id, date, region) DO UPDATE SET
                total_views = video_analytics.total_views + EXCLUDED.total_views,
                metrics = video_analytics.metrics || EXCLUDED.metrics,
                updated_at = now()
        """, video_id, date.today(), region, total_views, json.dumps(metrics))


# Record UAE analytics with language mix data
await record_regional_analytics(
    pool,
    video_id="abc123",
    region="AE",
    total_views=42000,
    metrics={
        "completion_rate": 0.54,
        "device": {"mobile": 0.83, "desktop": 0.14},
        "language_mix": {"ar": 0.48, "en": 0.28, "hi": 0.15}
    }
)

The JSONB approach means that when we want to track a new metric — like subtitle language preference for Czech viewers or Ramadan-period viewing patterns for UAE — we simply start including it in the payload. No ALTER TABLE, no migration, no deployment coordination.

For a platform like TrendVidStream serving 8 culturally distinct regions, that flexibility is essential.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.

PostgreSQL JSONB for Tracking European Video Trends

ahmet gedik — Thu, 09 Apr 2026 18:00:01 +0000

Why JSONB for Regional Analytics

At ViralVidVault, each video has different analytics depending on which European region it trends in. Polish engagement rates, Dutch device breakdowns, Scandinavian watch-time patterns — this data has variable shape and evolves as we add new metrics. PostgreSQL JSONB handles this perfectly.

The Schema

CREATE TABLE video_trends (
    id SERIAL PRIMARY KEY,
    video_id TEXT NOT NULL REFERENCES videos(id),
    snapshot_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    total_views BIGINT NOT NULL,
    regional_data JSONB NOT NULL DEFAULT '{}',
    engagement JSONB NOT NULL DEFAULT '{}'
);

CREATE INDEX idx_trends_video ON video_trends(video_id, snapshot_at DESC);
CREATE INDEX idx_trends_regional ON video_trends USING GIN(regional_data);
CREATE INDEX idx_trends_engagement ON video_trends USING GIN(engagement);

A typical regional_data value:

{
  "PL": {"views": 45000, "trending_rank": 3, "category": "Music"},
  "NL": {"views": 12000, "trending_rank": 15, "category": "Entertainment"},
  "SE": {"views": 8200, "trending_rank": 22, "category": "Music"},
  "GB": {"views": 95000, "trending_rank": 8, "category": "Music"}
}

And engagement:

{
  "like_rate": 0.052,
  "comment_rate": 0.011,
  "avg_watch_seconds": 156,
  "devices": {"mobile": 0.58, "desktop": 0.35, "tablet": 0.07},
  "peak_hour_utc": 19
}

Querying Regional Data

Find videos trending in Poland with high engagement:

SELECT v.title, t.regional_data -> 'PL' ->> 'views' AS pl_views,
       t.engagement ->> 'like_rate' AS like_rate
FROM video_trends t
JOIN videos v ON v.id = t.video_id
WHERE t.regional_data ? 'PL'
  AND (t.engagement ->> 'like_rate')::float > 0.05
ORDER BY (t.regional_data -> 'PL' ->> 'views')::int DESC
LIMIT 20;

The ? operator checks if the key 'PL' exists. The ->> operator extracts a text value. Cast to float or int for numeric comparisons.

Aggregate views by region across all videos:

SELECT region.key AS region,
       COUNT(*) AS video_count,
       SUM((region.value ->> 'views')::bigint) AS total_views,
       AVG((region.value ->> 'trending_rank')::int) AS avg_rank
FROM video_trends t,
     jsonb_each(t.regional_data) AS region
WHERE t.snapshot_at > NOW() - INTERVAL '24 hours'
GROUP BY region.key
ORDER BY total_views DESC;

jsonb_each unpacks the JSONB object into key-value rows. This lets you aggregate across all regions in a single query — something that would require separate columns (views_pl, views_nl, views_se...) without JSONB.

Updating JSONB Incrementally

When new trending data arrives, update specific regions without overwriting the entire object:

-- Add or update a single region
UPDATE video_trends
SET regional_data = jsonb_set(
    regional_data,
    '{NO}',
    '{"views": 5400, "trending_rank": 18, "category": "Nature"}'::jsonb
)
WHERE video_id = 'abc123'
  AND snapshot_at = (SELECT MAX(snapshot_at) FROM video_trends WHERE video_id = 'abc123');

-- Merge new engagement data
UPDATE video_trends
SET engagement = engagement || '{"share_rate": 0.023}'::jsonb
WHERE video_id = 'abc123';

The || operator merges objects. Existing keys are updated, new keys are added. No need to read-modify-write in application code.

Time-Series Analytics for the Dashboard

ViralVidVault admin dashboard shows trends over time:

SELECT
    date_trunc('day', snapshot_at) AS day,
    jsonb_object_agg(
        region.key,
        region.value ->> 'views'
    ) AS views_by_region
FROM video_trends t,
     jsonb_each(t.regional_data) AS region
WHERE video_id = 'abc123'
  AND snapshot_at > NOW() - INTERVAL '30 days'
GROUP BY day
ORDER BY day;

This returns one row per day with a JSONB object showing views per region — exactly what a chart library needs.

Performance Observations

On ~50,000 trend records with GIN indexes:

Key existence check (?): 1-2ms
Path extraction with filter: 3-5ms
jsonb_each aggregation (24h window): 10-20ms
Full cross-region aggregation: 30-60ms

JSONB is not as fast as dedicated columns for simple lookups, but the flexibility to add regions, metrics, and breakdowns without schema changes is invaluable for a platform tracking trends across 7 European markets.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

Async Video Metadata Pipeline with Python asyncio and aiohttp

ahmet gedik — Sat, 28 Mar 2026 05:00:01 +0000

Async Video Metadata Pipeline with Python asyncio and aiohttp

Fetching trending video metadata from 9 regions sequentially takes about 18 seconds. Switching to asyncio with aiohttp drops that to under 3 seconds. Here's the async pipeline I built for TopVideoHub.

Why asyncio for YouTube API Calls

YouTube Data API calls are I/O-bound — your code spends most of its time waiting for HTTP responses. That's the perfect use case for asyncio: while one coroutine waits for Japan's trending data, another fetches Korea's, another Taiwan's. No OS threads needed.

The Core Fetch Coroutine

import asyncio
import aiohttp
import logging
from dataclasses import dataclass
from typing import Optional

logger = logging.getLogger(__name__)

REGIONS = ["JP", "KR", "TW", "SG", "VN", "TH", "HK", "US", "GB"]
YOUTUBE_BASE = "https://www.googleapis.com/youtube/v3"


@dataclass
class VideoMetadata:
    video_id: str
    title: str
    channel_title: str
    region: str
    view_count: int
    thumbnail_url: str
    published_at: str
    category_id: int


async def fetch_region(
    session: aiohttp.ClientSession,
    region: str,
    api_key: str,
    semaphore: asyncio.Semaphore,
    max_results: int = 50,
) -> list[VideoMetadata]:
    """Fetch trending videos for a single region, respecting rate limits."""
    async with semaphore:  # Only N concurrent requests allowed
        url = f"{YOUTUBE_BASE}/videos"
        params = {
            "part": "snippet,statistics,contentDetails",
            "chart": "mostPopular",
            "regionCode": region,
            "maxResults": max_results,
            "key": api_key,
        }
        try:
            timeout = aiohttp.ClientTimeout(total=30)
            async with session.get(url, params=params, timeout=timeout) as resp:
                if resp.status == 429:
                    logger.warning(f"[{region}] Rate limited, retrying after backoff")
                    await asyncio.sleep(5)
                    return []  # Caller can retry
                resp.raise_for_status()
                data = await resp.json()
                videos = [_parse_item(item, region) for item in data.get("items", [])]
                logger.info(f"[{region}] Fetched {len(videos)} videos")
                return videos
        except aiohttp.ClientError as e:
            logger.error(f"[{region}] HTTP error: {e}")
            return []
        except asyncio.TimeoutError:
            logger.error(f"[{region}] Request timed out")
            return []


def _parse_item(item: dict, region: str) -> VideoMetadata:
    snippet = item["snippet"]
    stats = item.get("statistics", {})
    thumbs = snippet.get("thumbnails", {})
    thumb_url = ""
    for quality in ("maxres", "standard", "high", "medium", "default"):
        if quality in thumbs:
            thumb_url = thumbs[quality]["url"]
            break
    return VideoMetadata(
        video_id=item["id"],
        title=snippet["title"],
        channel_title=snippet.get("channelTitle", ""),
        region=region,
        view_count=int(stats.get("viewCount", 0)),
        thumbnail_url=thumb_url,
        published_at=snippet.get("publishedAt", ""),
        category_id=int(snippet.get("categoryId", 0)),
    )

Gathering All Regions

async def fetch_all_regions(
    api_key: str,
    regions: list[str] = REGIONS,
    max_concurrent: int = 5,
) -> dict[str, list[VideoMetadata]]:
    """Fetch all regions concurrently with controlled parallelism."""
    semaphore = asyncio.Semaphore(max_concurrent)

    connector = aiohttp.TCPConnector(
        limit=10,           # Max total simultaneous connections
        ttl_dns_cache=300,  # Cache DNS lookups for 5 minutes
    )

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            fetch_region(session, region, api_key, semaphore)
            for region in regions
        ]
        # gather() runs all tasks concurrently, collects results
        results = await asyncio.gather(*tasks, return_exceptions=True)

    output = {}
    for region, result in zip(regions, results):
        if isinstance(result, Exception):
            logger.error(f"[{region}] Task exception: {result}")
            output[region] = []
        else:
            output[region] = result

    return output

Handling CJK Titles

Japanese, Korean, and Chinese titles from the API come back as proper Unicode. Python 3 handles this natively, but normalize before storing to ensure consistent representation:

import unicodedata

def normalize_cjk_title(title: str) -> str:
    """NFC normalization for consistent CJK Unicode storage."""
    # NFC: composed form (e.g. U+00E9 rather than e + combining acute)
    title = unicodedata.normalize("NFC", title)
    # Remove control characters but preserve all CJK scripts
    return "".join(
        ch for ch in title
        if unicodedata.category(ch) not in ("Cc", "Cs")
    ).strip()

Async Database Writes

Pair async fetching with async database writes for maximum throughput:

import asyncpg

async def bulk_upsert(pool: asyncpg.Pool, videos: list[VideoMetadata]) -> int:
    """Bulk upsert using executemany for efficient batch writes."""
    async with pool.acquire() as conn:
        await conn.executemany("""
            INSERT INTO videos
                (video_id, title, channel_title, region, view_count,
                 thumbnail_url, published_at, category_id)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            ON CONFLICT (video_id, region) DO UPDATE SET
                view_count = EXCLUDED.view_count,
                fetched_at = now()
        """, [
            (v.video_id, normalize_cjk_title(v.title), v.channel_title,
             v.region, v.view_count, v.thumbnail_url, v.published_at, v.category_id)
            for v in videos
        ])
    return len(videos)


async def main():
    import os
    api_key = os.environ["YOUTUBE_API_KEY"]
    db_url = os.environ["DATABASE_URL"]

    pool = await asyncpg.create_pool(db_url, min_size=2, max_size=5)

    all_videos = await fetch_all_regions(api_key)
    flat_videos = [v for videos in all_videos.values() for v in videos]

    inserted = await bulk_upsert(pool, flat_videos)
    logger.info(f"Upserted {inserted} videos from {len(all_videos)} regions")

    await pool.close()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

Performance Results

For TopVideoHub's 9 Asia-Pacific regions:

Approach	Time	Notes
Sequential requests	~18s	Simple loop
asyncio semaphore=3	~6s	3 concurrent
asyncio semaphore=9	~2.5s	All parallel

The semaphore value of 3-5 is the sweet spot — enough parallelism to be fast, few enough concurrent connections to avoid triggering YouTube's rate limiting. The return_exceptions=True argument to gather() is critical: without it, a single region failure would cancel all other pending requests.

This article is part of the Building TopVideoHub series. Check out TopVideoHub to see these techniques in action.

PostgreSQL JSONB for Video Analytics Dashboards

ahmet gedik — Fri, 27 Mar 2026 22:00:01 +0000

The Problem with Rigid Schemas for Analytics

Video analytics data is inherently messy. One day you are tracking view counts by region, the next you need device breakdowns, then watch-time percentiles. Adding a column for every new metric leads to a table with 50 nullable columns and constant migrations. At DailyWatch, we solved this with PostgreSQL's JSONB column type.

Schema Design

We keep structured data in normal columns and flexible analytics in JSONB:

CREATE TABLE video_analytics (
    id SERIAL PRIMARY KEY,
    video_id TEXT NOT NULL REFERENCES videos(video_id),
    captured_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    views INTEGER NOT NULL,
    likes INTEGER NOT NULL,
    metrics JSONB NOT NULL DEFAULT '{}'
);

The metrics column stores everything that might change shape over time:

{
  "regions": {"US": 45000, "GB": 12000, "DE": 8500, "FR": 6200},
  "devices": {"mobile": 0.62, "desktop": 0.31, "tablet": 0.07},
  "watch_time": {"avg_seconds": 142, "p50": 98, "p95": 340},
  "engagement": {"like_rate": 0.034, "comment_rate": 0.008},
  "traffic_sources": {"search": 0.35, "suggested": 0.42, "external": 0.23}
}

Indexing JSONB for Fast Queries

A GIN index on the JSONB column lets PostgreSQL search inside the JSON efficiently:

-- General-purpose GIN index
CREATE INDEX idx_analytics_metrics ON video_analytics USING GIN(metrics);

-- Targeted index for a specific path (faster, smaller)
CREATE INDEX idx_analytics_regions ON video_analytics
    USING GIN((metrics -> 'regions'));

The general GIN index supports containment (@>) and existence (?) operators. The path-specific index is smaller and faster if you know which keys you query most.

Querying JSONB Data

Here is how we extract analytics for the admin dashboard at DailyWatch:

Get regional view breakdown:

SELECT video_id,
       metrics -> 'regions' ->> 'US' AS us_views,
       metrics -> 'regions' ->> 'DE' AS de_views,
       metrics -> 'watch_time' ->> 'avg_seconds' AS avg_watch
FROM video_analytics
WHERE captured_at > NOW() - INTERVAL '24 hours'
ORDER BY views DESC
LIMIT 20;

Find videos where mobile traffic exceeds 70%:

SELECT v.title, a.metrics -> 'devices' ->> 'mobile' AS mobile_share
FROM video_analytics a
JOIN videos v ON v.video_id = a.video_id
WHERE (a.metrics -> 'devices' ->> 'mobile')::float > 0.70
ORDER BY a.captured_at DESC;

Aggregate regional views across all videos:

SELECT
    key AS region,
    SUM(value::integer) AS total_views
FROM video_analytics,
     jsonb_each_text(metrics -> 'regions')
WHERE captured_at > NOW() - INTERVAL '7 days'
GROUP BY key
ORDER BY total_views DESC;

The jsonb_each_text function unpacks a JSONB object into rows — incredibly useful for aggregation across dynamic keys.

Appending to JSONB Without Full Replacement

PostgreSQL's jsonb_set function updates specific paths without replacing the entire object:

-- Add a new traffic source
UPDATE video_analytics
SET metrics = jsonb_set(metrics, '{traffic_sources,direct}', '0.15')
WHERE video_id = 'abc123';

-- Merge new data into existing object
UPDATE video_analytics
SET metrics = metrics || '{"cdn_cost": 0.0023}'::jsonb
WHERE video_id = 'abc123';

The || operator merges objects. Existing keys are overwritten, new keys are added. This is how we incrementally enrich analytics records as new data arrives.

Building a Time-Series View

For dashboard charts, we need metrics over time:

SELECT
    date_trunc('hour', captured_at) AS hour,
    AVG(views) AS avg_views,
    AVG((metrics -> 'watch_time' ->> 'avg_seconds')::float) AS avg_watch_time,
    AVG((metrics -> 'devices' ->> 'mobile')::float) AS mobile_share
FROM video_analytics
WHERE video_id = 'abc123'
  AND captured_at > NOW() - INTERVAL '7 days'
GROUP BY hour
ORDER BY hour;

This gives us hourly data points for charting view trends, watch time, and device mix — all from a single JSONB column.

When NOT to Use JSONB

JSONB is not a replacement for proper schema design. We use normal columns for data we always query (video_id, views, likes, captured_at) and JSONB for data that is optional, variable-shape, or infrequently filtered.

Bad use of JSONB: storing video_id or title inside JSON when you need to JOIN on them.
Good use of JSONB: storing analytics breakdowns that vary by source and evolve over time.

Performance Notes

On our dataset of ~50,000 analytics records with GIN indexes:

Path extraction (->>) queries: 1-3ms
Containment (@>) queries: 2-5ms
jsonb_each_text aggregation: 10-20ms
Full-table JSONB aggregation: 40-80ms

The flexibility of JSONB let us iterate on our analytics dashboard without a single migration after the initial table creation.

This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.

PostgreSQL Full-Text Search for Multi-Script Video Metadata

ahmet gedik — Fri, 27 Mar 2026 12:00:00 +0000

PostgreSQL Full-Text Search for Multi-Script Video Metadata

TrendVidStream aggregates trending video content from 8 regions spanning the UAE (Arabic), Nordic countries (Danish, Finnish), and Central Europe (Czech, Belgian French/Dutch). Each region brings different script requirements to full-text search — and PostgreSQL handles most of them well.

Here's what I learned building FTS for a multi-region, multi-script video platform.

Setting Up Weighted tsvectors

For video metadata, titles matter more than descriptions. PostgreSQL's weight system (A through D) lets you encode this in the index:

CREATE TABLE videos (
    id          BIGSERIAL PRIMARY KEY,
    video_id    TEXT UNIQUE NOT NULL,
    title       TEXT NOT NULL,
    channel_title TEXT,
    description TEXT,
    region      TEXT NOT NULL,
    category_id INTEGER,
    view_count  BIGINT DEFAULT 0,
    language    TEXT,  -- 'ar', 'da', 'fi', 'cs', 'nl', 'fr', 'en'
    published_at TIMESTAMPTZ,
    fetched_at  TIMESTAMPTZ DEFAULT now(),
    -- Weighted vector: title (A) > channel (B) > description (C)
    search_vec  TSVECTOR GENERATED ALWAYS AS (
        setweight(to_tsvector('simple', coalesce(title, '')), 'A') ||
        setweight(to_tsvector('simple', coalesce(channel_title, '')), 'B') ||
        setweight(to_tsvector('simple', coalesce(left(description, 500), '')), 'C')
    ) STORED
);

CREATE INDEX idx_videos_search ON videos USING GIN(search_vec);
CREATE INDEX idx_videos_region_cat ON videos(region, category_id);

RTL Arabic Text Considerations

UAE content on TrendVidStream includes Arabic titles. Arabic is RTL but PostgreSQL's FTS tokenizer treats it as UTF-8 text regardless of direction — the direction is a rendering concern, not a storage concern.

The simple configuration lowercases and splits on whitespace, which works reasonably for Arabic since Arabic words are space-separated unlike CJK languages:

-- Arabic search works with 'simple' config
SELECT video_id, title
FROM videos
WHERE search_vec @@ to_tsquery('simple', 'موسيقى')
  AND region = 'AE';

For better Arabic FTS, you'd want an Arabic dictionary with root-word stemming. But for a video discovery platform where users search for exact artist names and song titles, simple works well enough.

Language-Specific Configurations

For Western European languages in the TrendVidStream region set, PostgreSQL has built-in configurations:

-- Finnish has an excellent built-in stemmer
SELECT to_tsvector('finnish', 'musiikkivideo suosittu');
-- Stems to: 'musiikkivideo' + 'suosittu' (root forms)

-- Danish stems work well too
SELECT to_tsvector('danish', 'populær musik video');

-- For multi-language search across all regions, use 'simple'
-- to avoid language-specific stemming artifacts
SELECT to_tsvector('simple', 'popular music');

The tradeoff: language-specific configs give better search for single-language queries (searching "musik" in Danish finds "musikvideo"), but mix poorly across languages. I use simple for the cross-region search index.

GIN Index Strategies for Multi-Region Queries

-- Partial index for each major region (smaller, faster for region-filtered searches)
CREATE INDEX idx_videos_search_ae ON videos USING GIN(search_vec)
    WHERE region = 'AE';

CREATE INDEX idx_videos_search_fi ON videos USING GIN(search_vec)
    WHERE region = 'FI';

-- Full index for cross-region search
CREATE INDEX idx_videos_search_all ON videos USING GIN(search_vec);

Querying with ts_rank and Region Filtering

-- Search across Nordic regions (FI + DK) with relevance ranking
SELECT
    video_id,
    title,
    region,
    ts_rank(search_vec, query, 32) AS rank
FROM videos,
     to_tsquery('simple', plainto_tsquery('simple', $1)::text) AS query
WHERE search_vec @@ query
  AND region IN ('FI', 'DK')
ORDER BY rank DESC
LIMIT 20;

-- Multi-region trending for Central European markets
SELECT
    video_id,
    title,
    region,
    view_count
FROM videos
WHERE region IN ('CZ', 'BE', 'CH')
  AND fetched_at > now() - interval '24 hours'
ORDER BY view_count DESC
LIMIT 50;

Python Batch Indexing

import psycopg2
from psycopg2.extras import execute_values

def bulk_index_videos(conn, videos: list[dict]) -> None:
    """Bulk upsert; GENERATED ALWAYS AS updates search_vec automatically."""
    with conn.cursor() as cur:
        execute_values(cur, """
            INSERT INTO videos (video_id, title, channel_title, description,
                                region, category_id, view_count, language)
            VALUES %s
            ON CONFLICT (video_id) DO UPDATE SET
                title = EXCLUDED.title,
                view_count = EXCLUDED.view_count,
                fetched_at = now()
        """, [
            (v['video_id'], v['title'], v.get('channel_title', ''),
             v.get('description', '')[:500], v['region'],
             v.get('category_id', 0), v.get('view_count', 0),
             v.get('language', 'en'))
            for v in videos
        ])
    conn.commit()

The GENERATED ALWAYS AS STORED column means search_vec updates automatically on every upsert — no triggers, no manual index maintenance.

Regulatory Considerations for Multi-Region Content

One non-technical challenge unique to TrendVidStream's region set: content that's available in one region may be restricted in another. Finnish content regulations, UAE content standards, and Czech licensing rules are all different.

From a database perspective, this means storing region availability explicitly and filtering at query time, not assuming cross-region availability.

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.

PostgreSQL Full-Text Search for Multi-Language Video Content

ahmet gedik — Fri, 27 Mar 2026 04:00:01 +0000

The Multi-Language Search Challenge

When your video platform indexes content from Poland, the Netherlands, Sweden, Norway, and Austria, a single-language search engine will not cut it. At ViralVidVault, we needed search that correctly stems Polish words, handles Swedish compound nouns, and still works for English queries. PostgreSQL's built-in full-text search delivered all of this without adding Elasticsearch to the stack.

Setting Up Language-Aware tsvectors

The key is storing a pre-computed tsvector column with the appropriate language dictionary per video:

ALTER TABLE videos ADD COLUMN search_vector tsvector;
ALTER TABLE videos ADD COLUMN content_language TEXT DEFAULT 'english';

-- Map regions to PostgreSQL dictionaries
-- PL=simple (no Polish dict in default PG), NL=dutch, SE=swedish, NO=norwegian, AT=german
UPDATE videos SET content_language = CASE
    WHEN region = 'NL' THEN 'dutch'
    WHEN region = 'SE' THEN 'swedish'
    WHEN region = 'NO' THEN 'norwegian'
    WHEN region IN ('AT', 'DE') THEN 'german'
    ELSE 'english'
END;

UPDATE videos SET search_vector =
    setweight(to_tsvector(content_language::regconfig, coalesce(title, '')), 'A') ||
    setweight(to_tsvector('simple', coalesce(channel_title, '')), 'B') ||
    setweight(to_tsvector(content_language::regconfig, coalesce(description, '')), 'C');

The 'simple' dictionary for channel names is intentional — channel names are proper nouns that should not be stemmed. Using the language-specific dictionary for titles and descriptions means "löpning" (Swedish for "running") correctly matches "löpa" and "springer" (Dutch/German for "runner") gets stemmed properly.

GIN Index and Trigger

CREATE INDEX idx_videos_fts ON videos USING GIN(search_vector);

CREATE OR REPLACE FUNCTION videos_search_update() RETURNS trigger AS $$
BEGIN
    NEW.search_vector :=
        setweight(to_tsvector(
            COALESCE(NEW.content_language, 'english')::regconfig,
            coalesce(NEW.title, '')
        ), 'A') ||
        setweight(to_tsvector('simple', coalesce(NEW.channel_title, '')), 'B') ||
        setweight(to_tsvector(
            COALESCE(NEW.content_language, 'english')::regconfig,
            coalesce(NEW.description, '')
        ), 'C');
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_videos_fts
    BEFORE INSERT OR UPDATE OF title, description, channel_title, content_language
    ON videos
    FOR EACH ROW EXECUTE FUNCTION videos_search_update();

The trigger fires on relevant column changes, keeping the search vector in sync without application code.

Cross-Language Search Queries

Users search in their own language. A Dutch user searching for "muziek" should find Dutch music videos. An English user searching "music" should find English-tagged content. Here is the query we use:

-- Search with automatic language detection
SELECT video_id, title, channel_title, region,
       ts_rank(search_vector, query) AS rank
FROM videos,
     to_tsquery('simple', 'music') AS query
WHERE search_vector @@ query
ORDER BY rank DESC, views DESC
LIMIT 20;

Using the simple configuration for the query means it matches against all language vectors without stemming the search term. For language-specific searches, pass the dictionary:

-- Dutch-specific search: "muziek" stems correctly
SELECT video_id, title, ts_rank(search_vector, query) AS rank
FROM videos,
     to_tsquery('dutch', 'muziek') AS query
WHERE search_vector @@ query
ORDER BY rank DESC;

Comparison with SQLite FTS5

ViralVidVault started on SQLite FTS5. Here is what drove the migration:

Feature	SQLite FTS5	PostgreSQL FTS
Language dictionaries	English only (default)	30+ built-in
Weighted ranking	Manual BM25 config	Native A/B/C/D weights
Compound word handling	None	Language-aware splitting
Diacritics	Manual unaccent	Dictionary-handled
Index updates	Rebuild or merge	Trigger-based, live

For a European-focused platform, the multi-language support alone justified the switch. Swedish compound nouns like "fotbollsspelare" (football player) get tokenized correctly with the Swedish dictionary. FTS5 treats it as one opaque token.

Performance on European Content

With ~12,000 videos across 7 regions indexed with GIN:

Single-term search (any language): 1-3ms
Multi-term AND query: 3-5ms
Cross-language search (simple config): 2-4ms
Language-specific stemmed search: 2-4ms
Full reindex after bulk insert: ~1.5 seconds

The GIN index adds about 3MB of storage but makes the difference between a table scan and an instant lookup.

This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.

DEV Community: ahmet gedik

High-Performance Video API Server with Go net/http

High-Performance Video API Server with Go net/http

Why Go for This Use Case

Core Types

HTTP Handler

Middleware Chaining

Main Server

Video Platform API with Python FastAPI

Building a Video API from Scratch

Project Structure

Pydantic Models

Async Database Layer

API Endpoints

Dependency Injection for Caching

Running and Testing

Async Video Pipeline with Python asyncio for Multi-Region Fetching

Async Video Pipeline with Python asyncio for Multi-Region Fetching

Core Fetch Function

Concurrent Fetch for All Regions

Handling Arabic Text from UAE

Processing Pipeline with Language Detection

Async Video Processing Pipeline with Python for European Content

Processing Video Data from 7 European Regions

The Fetcher with Regional Awareness

Deduplication with Region Merging

Running the Full Pipeline

Performance Numbers

Building a Video Platform API with Python FastAPI

Building a Video Platform API with Python FastAPI

Project Setup

Pydantic Models

Dependency Injection for Database

Trending Videos Router

Search Endpoint with PostgreSQL FTS

Application Entry Point

Async Video Processing Pipeline with Python asyncio

Why Async Matters for Video Metadata

The Basic Pattern

Adding Rate Limiting with a Semaphore

Building a Processing Pipeline

Error Handling Patterns

Performance Comparison

PostgreSQL JSONB for Multi-Region Video Analytics

PostgreSQL JSONB for Multi-Region Video Analytics

Schema Design

Inserting Rich Analytics

Analytics Queries for Platform Insights

Querying JSONB Arrays

Python Integration

PostgreSQL JSONB for Tracking European Video Trends

Why JSONB for Regional Analytics

The Schema

Querying Regional Data

Updating JSONB Incrementally

Time-Series Analytics for the Dashboard

Performance Observations

Async Video Metadata Pipeline with Python asyncio and aiohttp

Async Video Metadata Pipeline with Python asyncio and aiohttp

Why asyncio for YouTube API Calls

The Core Fetch Coroutine

Gathering All Regions

Handling CJK Titles

Async Database Writes

Performance Results

PostgreSQL JSONB for Video Analytics Dashboards

The Problem with Rigid Schemas for Analytics

Schema Design

Indexing JSONB for Fast Queries

Querying JSONB Data

Appending to JSONB Without Full Replacement

Building a Time-Series View

When NOT to Use JSONB

Performance Notes

PostgreSQL Full-Text Search for Multi-Script Video Metadata

PostgreSQL Full-Text Search for Multi-Script Video Metadata

Setting Up Weighted tsvectors

RTL Arabic Text Considerations

Language-Specific Configurations

GIN Index Strategies for Multi-Region Queries