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

Trending Videos Router

from fastapi import APIRouter, Depends, Query, HTTPException
from fastapi_cache.decorator import cache
import asyncpg
from .database import get_conn
from .models import Region, TrendingResponse, VideoResponse

router = APIRouter(prefix="/trending", tags=["trending"])


@router.get("/{region}", response_model=TrendingResponse)
@cache(expire=3600)  # Cache for 1 hour at the API layer
async def get_trending(
    region: Region,  # Pydantic validates this is a valid region enum value
    category_id: int | None = Query(default=None, description="YouTube category ID filter"),
    limit: int = Query(default=50, ge=1, le=100),
    conn: asyncpg.Connection = Depends(get_conn),
):
    query = """
        SELECT v.video_id, v.title, v.channel_title, v.thumbnail_url,
               v.view_count, v.published_at, v.category_id,
               vr.region, vr.fetched_at
        FROM videos v
        JOIN video_regions vr ON vr.video_id = v.id
        WHERE vr.region = $1
    """
    params: list = [region.value]

    if category_id is not None:
        query += " AND v.category_id = $2"
        params.append(category_id)

    query += f" ORDER BY vr.trending_rank ASC NULLS LAST, v.view_count DESC LIMIT ${len(params)+1}"
    params.append(limit)

    rows = await conn.fetch(query, *params)
    if not rows:
        raise HTTPException(status_code=404, detail=f"No trending data for region {region.value}")

    videos = [VideoResponse(**dict(r)) for r in rows]
    return TrendingResponse(
        region=region,
        videos=videos,
        fetched_at=rows[0]["fetched_at"],
        total=len(videos),
    )

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.